使用Spark Scala在MongoDB中保存流式数据帧

import org.apache.spark.sql.{DataFrame, SparkSession}
import com.mongodb.spark.MongoSpark

val spark = SparkSession.builder()
  .appName("Save Streaming DataFrame to MongoDB")
  .config("spark.mongodb.input.uri", "mongodb://localhost/test.inputCollection")
  .config("spark.mongodb.output.uri", "mongodb://localhost/test.outputCollection")
  .getOrCreate()

val streamingDF: DataFrame = spark.readStream
  .format("socket")
  .option("host", "localhost")
  .option("port", 9999)
  .load()

val processedDF: DataFrame = streamingDF.selectExpr("value as data")

val query = processedDF.writeStream
  .outputMode("append")
  .format("mongo")
  .option("database", "test")
  .option("collection", "outputCollection")
  .start()