如何使用 spark 和 mongo Hadoop 连接器更新 mongo 集合中的多个字段？

Question

我正在使用 mongo hadoop 连接器将 mongo 与 spark 和 scala 连接起来。

我想使用 mongo 和 spark 更新（更新）mongo 集合的多个值。

我试过关注--

val query = new BasicBSONObject()
query.append("customerId", customerId)

val update = new BasicBSONObject()
val bson = new BasicBSONObject()
bson.put("minimumUtilization", some value)
bson.put("maximumUtilization", some value)
bson.put("averageUtilization", some value)
bson.put("customerId", customerId)

现在我如何使用上述对象使用 spark 和 mongo hadoop 连接器更新集合？

Answer 1

请看这里：https://github.com/mongodb/mongo-hadoop/wiki/Spark-Usage

使用 MongoUpdateWritable。 然后创建一个 MongoUpdateWritable(query,update,upsert,multiupdate,replace) 的 RDD

updates.map(f =>(null,f)).saveAsNewAPIHadoopFile(
      "file:///this-is-completely-unused",
      classOf[Object],
      classOf[MongoUpdateWritable],
      classOf[MongoOutputFormat[Object, MongoUpdateWritable]],
      outputConfig)