从 Spark 自定义 Mongo 中的写入操作

Question

考虑到以下情况，我如何使用 spark 向 mongo 写入数据：

1. 如果文档存在，只需使用更新的值更新匹配的字段，如果该字段不存在，则添加新字段。 （replaceDocument 参数如果 false 将更新匹配记录但不添加新的不匹配字段，如果设置为 true，我的旧字段可以得到丢失。）
2. 我想将数据字段保持为只读，例如有两个字段，first_load_date 和 updated_on。 first_load_date 永远不会改变，它是在 mongo 中创建记录的日期，而 updated_on 是添加新字段或替换旧字段的日期。
3. 如果文档不存在，请插入。

主要问题是 replaceDocument = True 将导致丢失较新行中不存在的旧字段，而 False 将处理匹配但现在较新的传入字段。

我正在使用 Mongo-Spark-Connector 2.4.1

df.write.format("mongo").mode("append").option("replaceDocument","true").option("database","db1").option("collection","my_collection").save()

Answer 1

我了解您在这里想要实现的目标： 你可以使用类似的东西：

(df
 .write
 .format("mongo")
 .mode("append")
 .option("ordered", "false")
 .option("replaceDocument", "false")
 .option("database", "db1")
 .option("collection", "my_collection")
 .save()
 )

设置为false 的replaceDocument 将帮助您保留旧记录并更新匹配的记录，同时您可以获得BulkWriteException，设置为false 的有序参数将有所帮助。