PySpark MongoDB 从 DataFrame 追加数组的所有元素

Question

我有一个如下所示的 MongoDB 集合：

{
    "_id" : { "customerName" : "Bob",  "customerPhone" : "123-456-7890"},
    "purchases": ["A", "B", "C", "D"]
}

基本上，_id 是一对关于客户的唯一键，而购买是客户购买的商品的数组。

我还有一个 PySpark DataFrame，我想将它推送到这个集合中，其中包含我想更新这个特定文档的信息。

df.write.format("com.mongodb.spark.sql.DefaultSource").mode("append") \
                .option("spark.mongodb.output.uri", "mongodb://localhost:27017/customer.purchases").save()

问题是，如果我要更新此文档，我想为 Bob 添加新购买，它只会在 purchases 中附加不存在的内容，而不是全部附加。

因此，我现在最终要做的是，我只需要调用 rdd.collect() 将整个内容转换为列表，而不是使用架构将其转换为 DataFrame。然后在检查密钥是否存在的同时将所有内容一一插入；当 RDD 的查询变大时，这会导致这部分变慢并且需要大量内存。

对于版本：

PySpark：2.2 MongoDB：3.0.15 Mongo Spark 连接器：2.2.1

如果我可以使用数据框将数组中的所有元素附加到 MongoDB 集合中，是否有人可以做些什么？ 另外，如果我有什么遗漏或其他我应该做的事情，请告诉我。 谢谢！

Answer 1

您需要更改文档的data models 或架构。这里的重要部分是 _id 关键字段。字段名称_id 保留用作主键；它的值在集合中必须是唯一的，是不可变的，并且可以是数组以外的任何类型。

在您的情况下，_id 字段的值是可变的，实际上这就是您要更新的内容。作为建议，您可能希望将其更改为：

{ "_id" : <unique identifier>
  "customerName" : "Bob",  
  "customerPhone" : "123-456-7890",
  "purchases": ["A", "B", "C", "D"]
}

您可以使用默认的_id 值ObjectId 作为唯一标识符。

在_id 字段上拥有唯一标识符后，我们来谈谈更新操作。从MongoDB Spark Connector v1.1+（当前版本2.2）开始，如果Dataframe在写入过程中包含_id字段，则数据将为upsert-ed。这意味着将更新具有相同 _id 值的任何现有文档，并且将插入集合中没有现有 _id 值的新文档。

奖金回合：

• 您还需要为purchases 字段找到更好的架构。具有未定义长度的数组长度可能会在将来产生问题。即鲍勃一年进行了 1000 次购买。

• 请更新您的MongoDB服务器版本（3.0.x版本是2015年的），目前稳定版本为3.4，下个月发布3.6。