Spark Mongo 连接器：在 MongoDB 连接中仅更新一个属性答案

【问题标题】：Spark Mongo connector: Upsert only one attribute in MongoDB connectionSpark Mongo 连接器：在 MongoDB 连接中仅更新一个属性
【发布时间】：2018-11-09 10:39:15
【问题描述】：

假设我有以下 Mongo 文档：

{
 "_id":1, 
 "age": 10,
 "foo": 20
}

以及以下 Spark DataFrame df：

_id | val
 1  | 'a'
 2  | 'b'

现在我想将数据框中的val 附加到 Mongo 文档中......

使用MongoDB Spark connector，我可以通过“_id”附加使用默认更新逻辑，这意味着如果Spark数据帧和Mongo文档中的“_id”匹配，Mongo连接器将不会创建新的文件，而是更新旧文件。

但是！更新的行为基本上类似于 replace - 如果我执行以下操作：

df
.write.format("com.mongodb.spark.sql.DefaultSource")
.mode("append")
.option('spark.mongodb.output.uri','mongodb://mongo_server:27017/testdb.test_collection')
.save()

集合将如下所示：

[   
    {
     "_id":1, 
     "val": 'a'
    },
   {
     "_id":2, 
     "val':'b' 
    }
]

我想得到这个：

[   
    {
     "_id":1, 
     "age": 10,
     "foo": 20
     "val": 'a'
    },
   {
     "_id":2, 
     "val':'b' 
    }
]

我的问题是：

有没有办法（某些选项）使 Spark 连接器的行为方式我想让它表现出来吗？
可以，我可以先从Mongo到Spark的文档，充实一下它们具有“val”属性并将其写入/附加回Mongo。这个操作的 I/O 是什么？是否满载（阅读所有文件，然后替换所有属性）还是有点聪明（比如阅读所有文档，但仅附加“val”属性，而不是替换整个文档）？

【问题讨论】：

标签： python mongodb apache-spark upsert

【解决方案1】：

有没有办法（某些选项）让 Spark 连接器按照我希望的方式运行？

是的，您可以将replaceDocument 设置为false。例如，在 Python 中使用MongoDB connector for Spark v2.2.2 和 Apache Spark v2.3：

df = sqlContext.read.format("com.mongodb.spark.sql.DefaultSource")
               .option("spark.mongodb.input.uri", "mongodb://host101:27017/dbName.collName").load()
df.first() 
> Row(_id=1.0, age=10.0, foo=20.0)

# Create a dataframe 
d = {'_id': [1, 2], 'val':['a', 'b']}
inputdf = pd.DataFrame(data=d) 
sparkdf = sqlContext.createDataFrame(inputdf)

# Write to Spark 
sparkdf.write.format("com.mongodb.spark.sql.DefaultSource")
             .mode("append").option("spark.mongodb.output.uri", "mongodb://host101:27017/dbName.collName")
             .option("replaceDocument", "false")
             .save()

# Result 
+---+----+----+---+
|_id| age| foo|val|
+---+----+----+---+
|1.0|10.0|20.0|  a|
|2.0|null|null|  b|
+---+----+----+---+

【讨论】：

是的，这正是我所需要的。非常感谢@Wan