Streaming Data With Apache Spark and MongoDB

1
{
2
  _id: ObjectId("624767546df0f7dd8783f300"),
3
  company_symbol: 'HSL',
4
  company_name: 'HUNGRY SYNDROME LLC',
5
  price: 45.74,
6
  tx_time: '2022-04-01T16:57:56Z'
7
}

1
from pyspark import SparkContext
2
from pyspark.streaming import StreamingContext
3
from pyspark.sql import SparkSession
4
from pyspark.sql.functions import *
5
6
spark = SparkSession.\
7
        builder.\
8
        appName("streamingExampleRead").\
9
        config('spark.jars.packages', 'org.mongodb.spark:mongo-spark-connector_2.12::10.1.1').\
10
        getOrCreate()
11
12
query=(spark.readStream.format("mongodb")
13
.option('spark.mongodb.connection.uri', '<CONNECTION STRING>')
14
    	.option('spark.mongodb.database', 'Stocks') \
15
    	.option('spark.mongodb.collection', 'StockData') \
16
.option('spark.mongodb.change.stream.publish.full.document.only','true') \
17
    	.option("forceDeleteTempCheckpointLocation", "true") \
18
    	.load())
19
20
query.printSchema()

1
query.isStreaming

1
query2=(query.writeStream \
2
  .outputMode("append") \
3
  .option("forceDeleteTempCheckpointLocation", "true") \
4
  .format("console") \
5
  .trigger(continuous="1 second")
6
  .start().awaitTermination());

1
from pyspark import SparkContext
2
from pyspark.streaming import StreamingContext
3
from pyspark.sql import SparkSession
4
from pyspark.sql import functions as F
5
from pyspark.sql.functions import *
6
from pyspark.sql.types import StructType,TimestampType, DoubleType, StringType, StructField
7
8
spark = SparkSession.\
9
        builder.\
10
        appName("streamingExampleWrite").\
11
        config('spark.jars.packages', 'org.mongodb.spark:mongo-spark-connector:10.1.1').\
12
        config('spark.jars.packages', 'org.apache.spark:spark-sql-kafka-0-10_2.12:3.0.0').\
13
        getOrCreate()
14
15
df = spark \
16
  .readStream \
17
  .format("kafka") \
18
  .option("startingOffsets", "earliest") \
19
  .option("kafka.bootstrap.servers", "KAFKA BROKER HOST HERE") \
20
  .option("subscribe", "stockdata.Stocks.StockData") \
21
  .load()
22
23
schemaStock = StructType([ \
24
    StructField("_id",StringType(),True), \
25
    StructField("company_name",StringType(), True), \
26
    StructField("company_symbol",StringType(), True), \
27
    StructField("price",StringType(), True), \
28
    StructField("tx_time",StringType(), True)])
29
30
schemaKafka = StructType([ \
31
    StructField("payload",StringType(),True)])

1
stockDF=df.selectExpr("CAST(value AS STRING)")

1
{
2
  "schema": {
3
    "type": "string",
4
    "optional": false
5
  },
6
  "payload": "{\"_id\": {\"$oid\": \"6249f8096df0f7dd8785d70a\"}, \"company_symbol\": \"GMI\", \"company_name\": \"GIDDY INNOVATIONS\", \"price\": 87.57, \"tx_time\": \"2022-04-03T15:39:53Z\"}"
7
}

1
stockDF=stockDF.select(from_json(col('value'),schemaKafka).alias("json_data")).selectExpr('json_data.*')

1
…
2
 {
3
    _id: ObjectId("624c6206e152b632f88a8ee2"),
4
    payload: '{"_id": {"$oid": "6249f8046df0f7dd8785d6f1"}, "company_symbol": "GMI", "company_name": "GIDDY MONASTICISM INNOVATIONS", "price": 87.62, "tx_time": "2022-04-03T15:39:48Z"}'
5
  }, …

1
stockDF=stockDF.select(from_json(col('payload'),schemaStock).alias("json_data2")).selectExpr('json_data2.*')

1
stockDF=stockDF.withColumn("tx_time",col("tx_time").cast("timestamp"))

1
dsw = (
2
  stockDF.writeStream
3
    .format("mongodb")
4
    .queryName("ToMDB")
5
    .option("checkpointLocation", "/tmp/pyspark7/")
6
    .option("forceDeleteTempCheckpointLocation", "true")
7
    .option('spark.mongodb.connection.uri', ‘<CONNECTION STRING>')
8
    .option('spark.mongodb.database', 'Stocks')
9
    .option('spark.mongodb.collection', 'Sink')
10
    .trigger(continuous="10 seconds")
11
    .outputMode("append")
12
    .start().awaitTermination());

1
query=(spark.readStream.format("mongodb").\
2
	option('spark.mongodb.connection.uri', '<<MONGODB CONNECTION STRING>>').\
3
	option('spark.mongodb.database', 'Stocks').\
4
	option('spark.mongodb.collection', 'StockData').\
5
	option('spark.mongodb.change.stream.publish.full.document.only','true').\
6
	option("forceDeleteTempCheckpointLocation", "true").\
7
	load())

1
dsw = (
2
  stockDF.writeStream
3
    .format("mongodb")
4
    .queryName("ToMDB")
5
    .option("checkpointLocation", "/tmp/pyspark7/")
6
    .option("forceDeleteTempCheckpointLocation", "true")
7
    .option('spark.mongodb.connection.uri', '<<MONGODB CONNECTION STRING>>')
8
    .option('spark.mongodb.database', 'Stocks')
9
    .option('spark.mongodb.collection', 'Sink')
10
    .trigger(continuous="10 seconds")
11
    .outputMode("append")
12
    .start().awaitTermination());