如何将 spark 结构流写入 mongodb 集合？答案

【问题标题】：How to write spark structure stream into mongodb collection?如何将 spark 结构流写入 mongodb 集合？
【发布时间】：2020-09-19 08:32:46
【问题描述】：

Spark 数据帧可以写入 mongodb 集合。参考-https://docs.mongodb.com/spark-connector/master/python/write-to-mongodb/

但是当尝试将 spark 结构流写入 mongodb 集合时，它不起作用。

您能否提出比在 udf 中使用 pymongo 代码更好的选择来实现这一点。

【问题讨论】：

你怎么知道“它不起作用。”？可以附上代码吗？

标签： mongodb apache-spark pyspark spark-structured-streaming

【解决方案1】：

使用foreachBatch sink 解决。 PFB 工作示例代码。

def write_mongo_row(df, epoch_id):
    mongoURL = "mongodb://XX.XX.XX.XX:27017/test.collection"
    df.write.format("mongo").mode("append").option("uri",mongoURL).save()
    pass

query=csvDF.writeStream.foreachBatch(write_mongo_row).start()
query.awaitTermination()

从How to use foreach or foreachBatch in PySpark to write to database?得到想法

【讨论】：

除了每次调用函数时初始化mongo连接之外，还有其他选择吗？

【解决方案2】：

分享一个替代解决方案，其中配置部分在一开始就被关注，而不是稍后在save 方法中处理配置（将配置与逻辑分开）。

def save(message: DataFrame):
    message.write \
        .format("mongo") \
        .mode("append") \
        .option("database", "db_name") \
        .option("collection", "collection_name") \
        .save()
    pass

spark: SparkSession = SparkSession \
    .builder \
    .appName("MyApp") \
    .config("spark.mongodb.input.uri", "mongodb://localhost:27017") \
    .config("spark.mongodb.output.uri", "mongodb://localhost:27017") \
    .config("spark.jars.packages", "org.mongodb.spark:mongo-spark-connector_2.12:3.0.1") \
    .master("local") \
    .getOrCreate()

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

query: StreamingQuery = df\
    .writeStream \
    .foreachBatch(save) \
    .start()

query.awaitTermination()

【讨论】：