Skip to content

S
SparkStreamMongoSave
Commit 56b53ee4 authored by Potharaju Peddi's avatar Potharaju Peddi
Browse files
Options

for sparkmongostream data save

parents
Hide whitespace changes
Inline Side-by-side
Showing with 530 additions and 0 deletions
.idea/.gitignore 0 → 100644
View file @ 56b53ee4
# Default ignored files
/shelf/
/workspace.xml
.idea/compiler.xml 0 → 100644
View file @ 56b53ee4
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="CompilerConfiguration">
<annotationProcessing>
<profile default="true" name="Default" enabled="true" />
<profile name="Maven default annotation processors profile" enabled="true">
<sourceOutputDir name="target/generated-sources/annotations" />
<sourceTestOutputDir name="target/generated-test-sources/test-annotations" />
<outputRelativeToContentRoot value="true" />
<module name="spark-poc" />
<module name="SparkStreamMongoSave" />
<module name="SparkStreamWithMongo" />
</profile>
</annotationProcessing>
</component>
</project>
\ No newline at end of file
.idea/jarRepositories.xml 0 → 100644
View file @ 56b53ee4
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="RemoteRepositoriesConfiguration">
<remote-repository>
<option name="id" value="central" />
<option name="name" value="Central Repository" />
<option name="url" value="https://repo.maven.apache.org/maven2" />
</remote-repository>
<remote-repository>
<option name="id" value="central" />
<option name="name" value="Maven Central repository" />
<option name="url" value="https://repo1.maven.org/maven2" />
</remote-repository>
<remote-repository>
<option name="id" value="jboss.community" />
<option name="name" value="JBoss Community repository" />
<option name="url" value="https://repository.jboss.org/nexus/content/repositories/public/" />
</remote-repository>
</component>
</project>
\ No newline at end of file
.idea/misc.xml 0 → 100644
View file @ 56b53ee4
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="ExternalStorageConfigurationManager" enabled="true" />
<component name="MavenProjectsManager">
<option name="originalFiles">
<list>
<option value="$PROJECT_DIR$/pom.xml" />
</list>
</option>
</component>
<component name="ProjectRootManager" version="2" languageLevel="JDK_1_8" default="true" project-jdk-name="1.8" project-jdk-type="JavaSDK">
<output url="file://$PROJECT_DIR$/out" />
</component>
</project>
\ No newline at end of file
.idea/uiDesigner.xml 0 → 100644
View file @ 56b53ee4
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="Palette2">
<group name="Swing">
<item class="com.intellij.uiDesigner.HSpacer" tooltip-text="Horizontal Spacer" icon="/com/intellij/uiDesigner/icons/hspacer.svg" removable="false" auto-create-binding="false" can-attach-label="false">
<default-constraints vsize-policy="1" hsize-policy="6" anchor="0" fill="1" />
</item>
<item class="com.intellij.uiDesigner.VSpacer" tooltip-text="Vertical Spacer" icon="/com/intellij/uiDesigner/icons/vspacer.svg" removable="false" auto-create-binding="false" can-attach-label="false">
<default-constraints vsize-policy="6" hsize-policy="1" anchor="0" fill="2" />
</item>
<item class="javax.swing.JPanel" icon="/com/intellij/uiDesigner/icons/panel.svg" removable="false" auto-create-binding="false" can-attach-label="false">
<default-constraints vsize-policy="3" hsize-policy="3" anchor="0" fill="3" />
</item>
<item class="javax.swing.JScrollPane" icon="/com/intellij/uiDesigner/icons/scrollPane.svg" removable="false" auto-create-binding="false" can-attach-label="true">
<default-constraints vsize-policy="7" hsize-policy="7" anchor="0" fill="3" />
</item>
<item class="javax.swing.JButton" icon="/com/intellij/uiDesigner/icons/button.svg" removable="false" auto-create-binding="true" can-attach-label="false">
<default-constraints vsize-policy="0" hsize-policy="3" anchor="0" fill="1" />
<initial-values>
<property name="text" value="Button" />
</initial-values>
</item>
<item class="javax.swing.JRadioButton" icon="/com/intellij/uiDesigner/icons/radioButton.svg" removable="false" auto-create-binding="true" can-attach-label="false">
<default-constraints vsize-policy="0" hsize-policy="3" anchor="8" fill="0" />
<initial-values>
<property name="text" value="RadioButton" />
</initial-values>
</item>
<item class="javax.swing.JCheckBox" icon="/com/intellij/uiDesigner/icons/checkBox.svg" removable="false" auto-create-binding="true" can-attach-label="false">
<default-constraints vsize-policy="0" hsize-policy="3" anchor="8" fill="0" />
<initial-values>
<property name="text" value="CheckBox" />
</initial-values>
</item>
<item class="javax.swing.JLabel" icon="/com/intellij/uiDesigner/icons/label.svg" removable="false" auto-create-binding="false" can-attach-label="false">
<default-constraints vsize-policy="0" hsize-policy="0" anchor="8" fill="0" />
<initial-values>
<property name="text" value="Label" />
</initial-values>
</item>
<item class="javax.swing.JTextField" icon="/com/intellij/uiDesigner/icons/textField.svg" removable="false" auto-create-binding="true" can-attach-label="true">
<default-constraints vsize-policy="0" hsize-policy="6" anchor="8" fill="1">
<preferred-size width="150" height="-1" />
</default-constraints>
</item>
<item class="javax.swing.JPasswordField" icon="/com/intellij/uiDesigner/icons/passwordField.svg" removable="false" auto-create-binding="true" can-attach-label="true">
<default-constraints vsize-policy="0" hsize-policy="6" anchor="8" fill="1">
<preferred-size width="150" height="-1" />
</default-constraints>
</item>
<item class="javax.swing.JFormattedTextField" icon="/com/intellij/uiDesigner/icons/formattedTextField.svg" removable="false" auto-create-binding="true" can-attach-label="true">
<default-constraints vsize-policy="0" hsize-policy="6" anchor="8" fill="1">
<preferred-size width="150" height="-1" />
</default-constraints>
</item>
<item class="javax.swing.JTextArea" icon="/com/intellij/uiDesigner/icons/textArea.svg" removable="false" auto-create-binding="true" can-attach-label="true">
<default-constraints vsize-policy="6" hsize-policy="6" anchor="0" fill="3">
<preferred-size width="150" height="50" />
</default-constraints>
</item>
<item class="javax.swing.JTextPane" icon="/com/intellij/uiDesigner/icons/textPane.svg" removable="false" auto-create-binding="true" can-attach-label="true">
<default-constraints vsize-policy="6" hsize-policy="6" anchor="0" fill="3">
<preferred-size width="150" height="50" />
</default-constraints>
</item>
<item class="javax.swing.JEditorPane" icon="/com/intellij/uiDesigner/icons/editorPane.svg" removable="false" auto-create-binding="true" can-attach-label="true">
<default-constraints vsize-policy="6" hsize-policy="6" anchor="0" fill="3">
<preferred-size width="150" height="50" />
</default-constraints>
</item>
<item class="javax.swing.JComboBox" icon="/com/intellij/uiDesigner/icons/comboBox.svg" removable="false" auto-create-binding="true" can-attach-label="true">
<default-constraints vsize-policy="0" hsize-policy="2" anchor="8" fill="1" />
</item>
<item class="javax.swing.JTable" icon="/com/intellij/uiDesigner/icons/table.svg" removable="false" auto-create-binding="true" can-attach-label="false">
<default-constraints vsize-policy="6" hsize-policy="6" anchor="0" fill="3">
<preferred-size width="150" height="50" />
</default-constraints>
</item>
<item class="javax.swing.JList" icon="/com/intellij/uiDesigner/icons/list.svg" removable="false" auto-create-binding="true" can-attach-label="false">
<default-constraints vsize-policy="6" hsize-policy="2" anchor="0" fill="3">
<preferred-size width="150" height="50" />
</default-constraints>
</item>
<item class="javax.swing.JTree" icon="/com/intellij/uiDesigner/icons/tree.svg" removable="false" auto-create-binding="true" can-attach-label="false">
<default-constraints vsize-policy="6" hsize-policy="6" anchor="0" fill="3">
<preferred-size width="150" height="50" />
</default-constraints>
</item>
<item class="javax.swing.JTabbedPane" icon="/com/intellij/uiDesigner/icons/tabbedPane.svg" removable="false" auto-create-binding="true" can-attach-label="false">
<default-constraints vsize-policy="3" hsize-policy="3" anchor="0" fill="3">
<preferred-size width="200" height="200" />
</default-constraints>
</item>
<item class="javax.swing.JSplitPane" icon="/com/intellij/uiDesigner/icons/splitPane.svg" removable="false" auto-create-binding="false" can-attach-label="false">
<default-constraints vsize-policy="3" hsize-policy="3" anchor="0" fill="3">
<preferred-size width="200" height="200" />
</default-constraints>
</item>
<item class="javax.swing.JSpinner" icon="/com/intellij/uiDesigner/icons/spinner.svg" removable="false" auto-create-binding="true" can-attach-label="true">
<default-constraints vsize-policy="0" hsize-policy="6" anchor="8" fill="1" />
</item>
<item class="javax.swing.JSlider" icon="/com/intellij/uiDesigner/icons/slider.svg" removable="false" auto-create-binding="true" can-attach-label="false">
<default-constraints vsize-policy="0" hsize-policy="6" anchor="8" fill="1" />
</item>
<item class="javax.swing.JSeparator" icon="/com/intellij/uiDesigner/icons/separator.svg" removable="false" auto-create-binding="false" can-attach-label="false">
<default-constraints vsize-policy="6" hsize-policy="6" anchor="0" fill="3" />
</item>
<item class="javax.swing.JProgressBar" icon="/com/intellij/uiDesigner/icons/progressbar.svg" removable="false" auto-create-binding="true" can-attach-label="false">
<default-constraints vsize-policy="0" hsize-policy="6" anchor="0" fill="1" />
</item>
<item class="javax.swing.JToolBar" icon="/com/intellij/uiDesigner/icons/toolbar.svg" removable="false" auto-create-binding="false" can-attach-label="false">
<default-constraints vsize-policy="0" hsize-policy="6" anchor="0" fill="1">
<preferred-size width="-1" height="20" />
</default-constraints>
</item>
<item class="javax.swing.JToolBar$Separator" icon="/com/intellij/uiDesigner/icons/toolbarSeparator.svg" removable="false" auto-create-binding="false" can-attach-label="false">
<default-constraints vsize-policy="0" hsize-policy="0" anchor="0" fill="1" />
</item>
<item class="javax.swing.JScrollBar" icon="/com/intellij/uiDesigner/icons/scrollbar.svg" removable="false" auto-create-binding="true" can-attach-label="false">
<default-constraints vsize-policy="6" hsize-policy="0" anchor="0" fill="2" />
</item>
</group>
</component>
</project>
\ No newline at end of file
.idea/vcs.xml 0 → 100644
View file @ 56b53ee4
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="VcsDirectoryMappings">
<mapping directory="$PROJECT_DIR$" vcs="Git" />
</component>
</project>
\ No newline at end of file
checkpointlocaion/streamingjob/commits/0 0 → 100644
View file @ 56b53ee4
v1
{"nextBatchWatermarkMs":0}
\ No newline at end of file
checkpointlocaion/streamingjob/commits/1 0 → 100644
View file @ 56b53ee4
v1
{"nextBatchWatermarkMs":0}
\ No newline at end of file
checkpointlocaion/streamingjob/commits/2 0 → 100644
View file @ 56b53ee4
v1
{"nextBatchWatermarkMs":0}
\ No newline at end of file
checkpointlocaion/streamingjob/commits/3 0 → 100644
View file @ 56b53ee4
v1
{"nextBatchWatermarkMs":0}
\ No newline at end of file
checkpointlocaion/streamingjob/commits/4 0 → 100644
View file @ 56b53ee4
v1
{"nextBatchWatermarkMs":0}
\ No newline at end of file
checkpointlocaion/streamingjob/commits/5 0 → 100644
View file @ 56b53ee4
v1
{"nextBatchWatermarkMs":0}
\ No newline at end of file
checkpointlocaion/streamingjob/commits/6 0 → 100644
View file @ 56b53ee4
v1
{"nextBatchWatermarkMs":0}
\ No newline at end of file
checkpointlocaion/streamingjob/commits/7 0 → 100644
View file @ 56b53ee4
v1
{"nextBatchWatermarkMs":0}
\ No newline at end of file
checkpointlocaion/streamingjob/commits/8 0 → 100644
View file @ 56b53ee4
v1
{"nextBatchWatermarkMs":0}
\ No newline at end of file
checkpointlocaion/streamingjob/commits/9 0 → 100644
View file @ 56b53ee4
v1
{"nextBatchWatermarkMs":0}
\ No newline at end of file
checkpointlocaion/streamingjob/metadata 0 → 100644
View file @ 56b53ee4
{"id":"bb56ce99-58e6-4071-931b-4904868d58f4"}
\ No newline at end of file
checkpointlocaion/streamingjob/offsets/0 0 → 100644
View file @ 56b53ee4
v1
{"batchWatermarkMs":0,"batchTimestampMs":1685334782622,"conf":{"spark.sql.streaming.stateStore.providerClass":"org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider","spark.sql.streaming.join.stateFormatVersion":"2","spark.sql.streaming.stateStore.compression.codec":"lz4","spark.sql.streaming.flatMapGroupsWithState.stateFormatVersion":"2","spark.sql.streaming.multipleWatermarkPolicy":"min","spark.sql.streaming.aggregation.stateFormatVersion":"2","spark.sql.shuffle.partitions":"200"}}
{"product-info-topic":{"0":8}}
\ No newline at end of file
checkpointlocaion/streamingjob/offsets/1 0 → 100644
View file @ 56b53ee4
v1
{"batchWatermarkMs":0,"batchTimestampMs":1685337385334,"conf":{"spark.sql.streaming.stateStore.providerClass":"org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider","spark.sql.streaming.join.stateFormatVersion":"2","spark.sql.streaming.stateStore.compression.codec":"lz4","spark.sql.streaming.flatMapGroupsWithState.stateFormatVersion":"2","spark.sql.streaming.multipleWatermarkPolicy":"min","spark.sql.streaming.aggregation.stateFormatVersion":"2","spark.sql.shuffle.partitions":"200"}}
{"product-info-topic":{"0":24}}
\ No newline at end of file
checkpointlocaion/streamingjob/offsets/2 0 → 100644
View file @ 56b53ee4
v1
{"batchWatermarkMs":0,"batchTimestampMs":1685451832717,"conf":{"spark.sql.streaming.stateStore.providerClass":"org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider","spark.sql.streaming.join.stateFormatVersion":"2","spark.sql.streaming.stateStore.compression.codec":"lz4","spark.sql.streaming.flatMapGroupsWithState.stateFormatVersion":"2","spark.sql.streaming.multipleWatermarkPolicy":"min","spark.sql.streaming.aggregation.stateFormatVersion":"2","spark.sql.shuffle.partitions":"200"}}
{"product-info-topic":{"0":16}}
\ No newline at end of file
checkpointlocaion/streamingjob/offsets/3 0 → 100644
View file @ 56b53ee4
v1
{"batchWatermarkMs":0,"batchTimestampMs":1685452064839,"conf":{"spark.sql.streaming.stateStore.providerClass":"org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider","spark.sql.streaming.join.stateFormatVersion":"2","spark.sql.streaming.stateStore.compression.codec":"lz4","spark.sql.streaming.flatMapGroupsWithState.stateFormatVersion":"2","spark.sql.streaming.multipleWatermarkPolicy":"min","spark.sql.streaming.aggregation.stateFormatVersion":"2","spark.sql.shuffle.partitions":"200"}}
{"product-info-topic":{"0":32}}
\ No newline at end of file
checkpointlocaion/streamingjob/offsets/4 0 → 100644
View file @ 56b53ee4
v1
{"batchWatermarkMs":0,"batchTimestampMs":1685452768778,"conf":{"spark.sql.streaming.stateStore.providerClass":"org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider","spark.sql.streaming.join.stateFormatVersion":"2","spark.sql.streaming.stateStore.compression.codec":"lz4","spark.sql.streaming.flatMapGroupsWithState.stateFormatVersion":"2","spark.sql.streaming.multipleWatermarkPolicy":"min","spark.sql.streaming.aggregation.stateFormatVersion":"2","spark.sql.shuffle.partitions":"200"}}
{"product-info-topic1":{"0":20}}
\ No newline at end of file
checkpointlocaion/streamingjob/offsets/5 0 → 100644
View file @ 56b53ee4
v1
{"batchWatermarkMs":0,"batchTimestampMs":1685453347332,"conf":{"spark.sql.streaming.stateStore.providerClass":"org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider","spark.sql.streaming.join.stateFormatVersion":"2","spark.sql.streaming.stateStore.compression.codec":"lz4","spark.sql.streaming.flatMapGroupsWithState.stateFormatVersion":"2","spark.sql.streaming.multipleWatermarkPolicy":"min","spark.sql.streaming.aggregation.stateFormatVersion":"2","spark.sql.shuffle.partitions":"200"}}
{"product-info-topic2":{"0":16}}
\ No newline at end of file
checkpointlocaion/streamingjob/offsets/6 0 → 100644
View file @ 56b53ee4
v1
{"batchWatermarkMs":0,"batchTimestampMs":1685453608276,"conf":{"spark.sql.streaming.stateStore.providerClass":"org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider","spark.sql.streaming.join.stateFormatVersion":"2","spark.sql.streaming.stateStore.compression.codec":"lz4","spark.sql.streaming.flatMapGroupsWithState.stateFormatVersion":"2","spark.sql.streaming.multipleWatermarkPolicy":"min","spark.sql.streaming.aggregation.stateFormatVersion":"2","spark.sql.shuffle.partitions":"200"}}
{"product-info-topic2":{"0":32}}
\ No newline at end of file
checkpointlocaion/streamingjob/offsets/7 0 → 100644
View file @ 56b53ee4
v1
{"batchWatermarkMs":0,"batchTimestampMs":1685454043522,"conf":{"spark.sql.streaming.stateStore.providerClass":"org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider","spark.sql.streaming.join.stateFormatVersion":"2","spark.sql.streaming.stateStore.compression.codec":"lz4","spark.sql.streaming.flatMapGroupsWithState.stateFormatVersion":"2","spark.sql.streaming.multipleWatermarkPolicy":"min","spark.sql.streaming.aggregation.stateFormatVersion":"2","spark.sql.shuffle.partitions":"200"}}
{"product-info-topic2":{"0":48}}
\ No newline at end of file
checkpointlocaion/streamingjob/offsets/8 0 → 100644
View file @ 56b53ee4
v1
{"batchWatermarkMs":0,"batchTimestampMs":1685454179185,"conf":{"spark.sql.streaming.stateStore.providerClass":"org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider","spark.sql.streaming.join.stateFormatVersion":"2","spark.sql.streaming.stateStore.compression.codec":"lz4","spark.sql.streaming.flatMapGroupsWithState.stateFormatVersion":"2","spark.sql.streaming.multipleWatermarkPolicy":"min","spark.sql.streaming.aggregation.stateFormatVersion":"2","spark.sql.shuffle.partitions":"200"}}
{"product-info-topic3":{"0":16}}
\ No newline at end of file
checkpointlocaion/streamingjob/offsets/9 0 → 100644
View file @ 56b53ee4
v1
{"batchWatermarkMs":0,"batchTimestampMs":1685518339393,"conf":{"spark.sql.streaming.stateStore.providerClass":"org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider","spark.sql.streaming.join.stateFormatVersion":"2","spark.sql.streaming.stateStore.compression.codec":"lz4","spark.sql.streaming.flatMapGroupsWithState.stateFormatVersion":"2","spark.sql.streaming.multipleWatermarkPolicy":"min","spark.sql.streaming.aggregation.stateFormatVersion":"2","spark.sql.shuffle.partitions":"200"}}
{"product-info-topic4":{"0":20}}
\ No newline at end of file
pom.xml 0 → 100644
View file @ 56b53ee4
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>org.example</groupId>
<artifactId>spark-poc</artifactId>
<version>1.0-SNAPSHOT</version>
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.12</artifactId>
<version>3.1.2</version>
</dependency>
<!-- Apache com.nisum.producer.SparkProducer SQL -->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.12</artifactId>
<version>3.1.2</version>
</dependency>
<!-- Apache com.nisum.producer.SparkProducer Streaming -->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.12</artifactId>
<version>3.1.2</version>
</dependency>
<!-- Kafka Clients -->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql-kafka-0-10_2.12</artifactId>
<version>3.1.2</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.8.0</version>
</dependency>
<!-- Kafka Streaming -->
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-streams</artifactId>
<version>2.8.0</version>
</dependency>
<dependency>
<groupId>org.mongodb.spark</groupId>
<artifactId>mongo-spark-connector_2.12</artifactId>
<version>3.0.1</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.26</version>
<scope>compile</scope>
</dependency>
</dependencies>
</project>
\ No newline at end of file
src/data/product-info.json 0 → 100644
View file @ 56b53ee4
[
{"productId":"101","productName":"mobile","productPrice":15000.00,"deliveryStatus":"ordered","date":"10-05-2023"},
{"productId":"102","productName":"charger","productPrice":25000.00,"deliveryStatus":"delivered","date":"15-05-2023"},
{"productId":"103","productName":"cooker","productPrice":35000.00,"deliveryStatus":"shipped","date":"25-05-2023"},
{"productId":"104","productName":"tv","productPrice":45000.00,"deliveryStatus":"shipped","date":"25-05-2023"},
{"productId":"105","productName":"laptop","productPrice":55000.00,"deliveryStatus":"delivered","date":"25-05-2023"},
{"productId":"106","productName":"bottle","productPrice":1500.00,"deliveryStatus":"ordered","date":"12-05-2023"},
{"productId":"107","productName":"mouse","productPrice":1000.00,"deliveryStatus":"delivered","date":"15-05-2023"},
{"productId":"108","productName":"keyboard","productPrice":500.00,"deliveryStatus":"shipped","date":"25-05-2023"},
{"productId":"109","productName":"table","productPrice":10000.00,"deliveryStatus":"shipped","date":"22-05-2023"},
{"productId":"110","productName":"phone","productPrice":12000.00,"deliveryStatus":"delivered","date":"23-05-2023"},
{"productId":"111","productName":"shop","productPrice":150000.00,"deliveryStatus":"ordered","date":"16-05-2023"},
{"productId":"112","productName":"byke","productPrice":75000.00,"deliveryStatus":"delivered","date":"18-05-2023"},
{"productId":"113","productName":"car","productPrice":20000.00,"deliveryStatus":"shipped","date":"27-05-2023"},
{"productId":"114","productName":"cycle","productPrice":33000.00,"deliveryStatus":"shipped","date":"28-05-2023"},
{"productId":"115","productName":"bed","productPrice":15000.00,"deliveryStatus":"delivered","date":"26-05-2023"},
{"productId":"116","productName":"chairs","productPrice":12000.00,"deliveryStatus":"ordered","date":"15-05-2023"},
{"productId":"117","productName":"server","productPrice":19555.00,"deliveryStatus":"delivered","date":"18-05-2023"},
{"productId":"118","productName":"van","productPrice":89200.00,"deliveryStatus":"shipped","date":"25-05-2023"},
{"productId":"119","productName":"designing","productPrice":14000.00,"deliveryStatus":"shipped","date":"25-05-2023"},
{"productId":"120","productName":"interier","productPrice":13000.00,"deliveryStatus":"delivered","date":"25-05-2023"}
]
src/main/java/io/nisum/spark/consumer/SparkStreamingConsumer.java 0 → 100644
View file @ 56b53ee4
package io.nisum.spark.consumer;
import com.mongodb.client.MongoClient;
import com.mongodb.client.MongoClients;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase;
import io.nisum.spark.entity.ProductInfo;
import lombok.extern.slf4j.Slf4j;
import org.apache.spark.sql.*;
import org.apache.spark.sql.streaming.OutputMode;
import org.apache.spark.sql.streaming.StreamingQuery;
import org.apache.spark.sql.types.StructType;
import org.bson.Document;
@Slf4j
public class SparkStreamingConsumer {
public static void main(String[] args) {
try {
SparkSession spark = SparkSession.builder()
.master("local[*]")
.appName("Structural Streaming")
.config("spark.mongodb.output.uri", "mongodb://localhost:27017/ProductsData.SparkStreamProductInfo")
.getOrCreate();
Dataset<Row> fromKafka = spark.readStream()
.format("kafka")
.option("kafka.bootstrap.servers", "127.0.0.1:9092")
.option("subscribe", "product-info-topic4")
.option("kafka.consumer.group.id", "localStream")
.option("failOnDataLoss", "false")
.option("startingOffsets", "earliest")
.load();
StructType schema = Encoders.bean(ProductInfo.class).schema();
Dataset<Row> dataset = fromKafka.selectExpr("cast(value as String) as message")
.select(functions.from_json(functions.col("message"), schema).as("data"))
.select("data.*")
.repartition(100);
Dataset<Row> filteredDataset = dataset.filter(functions.col("deliveryStatus").equalTo("shipped"));
Dataset<ProductInfo> ds = filteredDataset.as(Encoders.bean(ProductInfo.class));
StreamingQuery query = ds.writeStream().outputMode(OutputMode.Append())
.option("checkpointLocation", "checkpointlocaion/streamingjob")
.foreach(new ForeachWriter<ProductInfo>() {
private MongoClient mongoClient;
private MongoDatabase database;
private MongoCollection<Document> collection;
@Override
public boolean open(long partitionId, long epochId) {
mongoClient = MongoClients.create("mongodb://localhost:27017");
database = mongoClient.getDatabase("ProductsData");
collection = database.getCollection("SparkStreamProductInfo");
return true;
}
@Override
public void process(ProductInfo productInfo) {
Document query = new Document();
query.append("productId", productInfo.getProductId());
Document product = collection.find(query).first();
if( (product == null) || (product.isEmpty()) ) {
// Create a document with the object data
Document document = new Document();
document.append("productId", productInfo.getProductId());
document.append("productName", productInfo.getProductName());
document.append("productPrice", productInfo.getProductPrice());
document.append("deliveryStatus", productInfo.getDeliveryStatus());
document.append("date", productInfo.getDate());
// Insert the document into the collection
collection.insertOne(document);
}
}
@Override
public void close(Throwable errorOrNull) {
// Close the MongoDB connection
mongoClient.close();
}
}).start();
query.awaitTermination();
} catch (Exception e) {
e.printStackTrace();
log.error(e.getMessage());
}
}
}
src/main/java/io/nisum/spark/entity/ProductInfo.java 0 → 100644
View file @ 56b53ee4
package io.nisum.spark.entity;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@AllArgsConstructor
@NoArgsConstructor
@Data
public class ProductInfo {
String productId;
String productName;
double productPrice;
String deliveryStatus;
String date;
@Override
public String toString() {
return "ProductInfo{" +
"productId='" + productId + '\'' +
", productName='" + productName + '\'' +
", productPrice=" + productPrice +
", deliveryStatus='" + deliveryStatus + '\'' +
", date='" + date + '\'' +
'}';
}
}
src/main/java/io/nisum/spark/producer/SparkKafkaProducer.java 0 → 100644
View file @ 56b53ee4
package io.nisum.spark.producer;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.functions;
import java.util.HashMap;
import java.util.Map;
public class SparkKafkaProducer {
public static void main(String[] args) {
try {
SparkSession spark = SparkSession.builder()
.master("local[*]")
.appName("Produce Product Info to Kafka Topic")
.getOrCreate();
Dataset<Row> productInfoDf = spark.read()
.format("json")
.option("header", true )
.option("multiline", "true")
.load("src/data/product-info.json");
productInfoDf.printSchema();
Map<String, String> kafkaConfigMap = new HashMap<String, String>();
kafkaConfigMap.put("kafka.bootstrap.servers", "127.0.0.1:9092");
kafkaConfigMap.put("topic", "product-info-topic4");
productInfoDf
.select(
functions.to_json(
functions.struct(
productInfoDf.col("deliveryStatus"),
productInfoDf.col("productId"),
productInfoDf.col("productName"),
productInfoDf.col("productPrice"),
productInfoDf.col("date")
)
)
.alias("value")
).write()
.options(kafkaConfigMap)
.format("kafka")
.save();
} catch (Exception e) {
e.printStackTrace();
}
}
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment