apache-spark - PySpark MongoDB 附加 DataFrame 中数组的所有元素

Question

我有一个 MongoDB 集合，看起来像这样:

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

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

我还有一个 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。然后将所有内容一一插入，同时检查key是否存在；当 RDD 的查询变大时，这使得这部分速度变慢并且需要大量内存。

对于版本:

PySpark:2.2MongoDB:3.0.15Mongo 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"]
}

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

一旦您在 _id 字段上有了唯一标识符，我们就来讨论更新操作。从 MongoDB Spark Connector v1.1+(当前版本 2.2)开始，如果 Dataframe 在写入期间包含 _id 字段，则数据将为 upsert-ed 。这意味着具有相同 _id 值的任何现有文档都将被更新，并且集合中没有现有 _id 值的新文档将被插入。

奖励回合:

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

• 请更新您的MongoDB服务器版本(2015年版本为3.0.x)，当前稳定版本为3.4，下个月将发布3.6。