如何使用 MongoDB Sink 将 Kafka 与 Spark Structured Streaming 集成

Question

我正在尝试将 Kafka 与 Spark-Structured-Streaming 集成到 MongoDB Sink。如果我出错了，我需要帮助来纠正我的代码

集成了 Kafka-Spark 和 Spark-Mongo。现在尝试从 Kafka-Spark-Mongo

整合管道

import org.apache.spark.sql.streaming.Trigger
import com.mongodb.spark.sql._
import org.apache.spark.streaming._
import com.mongodb.spark._
import com.mongodb.spark.config._
import org.bson.Document

//Creates readStream from Kafka
val df = spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "10.170.172.45:9092, 10.180.172.46:9092, 10.190.172.100:9092")
.option("subscribe", "HANZO_TEST_P2_R2, TOPIC_WITH_COMP_P2_R2, TOPIC_WITH_COMP_P2_R2.DIT, TOPIC_WITHOUT_COMP_P2_R2.DIT")
.load()
//The read kafka streaming will need to converted to string from Binary format
val dfs = df.selectExpr("CAST(value AS STRING)").toDF()

//The below logic extracts data from _raw column and in the stream context it is "value"
val extractedDF = dfs
.withColumn("managed_server", regexp_extract($"value", "\[(.*?)\] \[(.*?)\]",2))
.withColumn("alert_summary", regexp_extract($"value", "\[(.*?)\] \[(.*?)\] \[(.*?)\]",3))
.withColumn("oracle_details", regexp_extract($"value", "\[(.*?)\] \[(.*?)\] \[(.*?)\] \[(.*?)\] \[(.*?)\]",5))
.withColumn("ecid", regexp_extract($"value", "(?<=ecid: )(.*?)(?=,)",1))
.withColumn("CompName",regexp_extract($"value",""".*(composite_name|compositename|composites|componentDN):\s+([a-zA-Z]+)""",2))
.withColumn("composite_name", col("value").contains("composite_name"))
.withColumn("compositename", col("value").contains("compositename"))
.withColumn("composites", col("value").contains("composites"))
.withColumn("componentDN", col("value").contains("componentDN"))

//The below logic filters any NULL values if found
val finalData = extractedDF.filter(
      col("managed_server").isNotNull &&
        col("alert_summary").isNotNull &&
        col("oracle_details").isNotNull &&
        col("ecid").isNotNull &&
        col("CompName").isNotNull &&
        col("composite_name").isNotNull &&
        col("compositename").isNotNull &&
        col("composites").isNotNull &&
        col("componentDN").isNotNull).toDF

val toMongo = MongoSpark.save(finalData.write.option("uri", "mongodb://hanzomdbuser:hanzomdbpswd@dstk8sd.com:27018/HANZO_MDB.Testing").mode("overwrite"))

//The Kafka stream should written and in this case we are writing it to console
val query = toMongo.writeStream
.outputMode("append")
.format("console")
.trigger(Trigger.ProcessingTime("20 seconds"))
.start()

query.awaitTermination()

我需要使用我的代码集成这三个框架，并且在 Spark 中处理后来自 Kafka 的所有流式处理结果需要保存在 MongoDB 中的集合中

Answer 1

您需要创建 Mongo 接收器，而不是您在示例中使用的 "console"。有一些有用的资源，例如：

https://github.com/mongodb/mongo-spark/blob/master/examples/src/test/scala/tour/SparkStructuredStreams.scala

和

https://github.com/holdenk/spark-structured-streaming-ml/blob/master/src/main/scala/com/high-performance-spark-examples/structuredstreaming/CustomSink.scala

和

https://learningfromdata.blog/2017/04/16/real-time-data-ingestion-with-apache-spark-structured-streaming-implementation/

如何使用 MongoDB Sink 将 Kafka 与 Spark Structured Streaming 集成

How to Integrate Kafka with Spark Structured Streaming with MongoDB Sink

scala

mongodb

apache-spark

apache-spark-sql

spark-structured-streaming