Spark MongoDB Connector如何实现文档已有数组的元素追加操作

原生Spark MongoDB Connector没有内置数组追加的配置项，但可以通过结合MongoDB原生驱动的自定义写入逻辑实现预期的数组合并效果。

你当前使用的append模式+replaceDocument=false配置，底层默认执行MongoDB的$set更新操作：

• 仅对DataFrame中存在、MongoDB文档中不存在的顶层字段做新增
• 对两边都存在的字段，无论字段类型是普通值还是数组，都会直接用DataFrame中的新值覆盖，所以才会出现field1数组被替换的情况。

通过foreachPartition批量调用MongoDB原生更新算子$push（可追加重复元素）或$addToSet（自动去重追加）实现数组追加，代码示例如下：

import com.mongodb.client.model.UpdateOptions
import com.mongodb.client.model.Updates.{pushEach, addToSetEach}
import org.bson.Document
import com.mongodb.spark.MongoConnector

// 需提前替换为你实际的Mongo连接uri
val mongoUri = "你的MongoDB连接地址"

// 待写入的DataFrame，必须包含_id字段和要追加的数组字段field1
dataframe.rdd.foreachPartition { partition =>
  // 分区级别复用连接，降低连接创建开销
  val connector = MongoConnector(Map("uri" -> mongoUri))
  connector.withCollectionDo { collection =>
    val batchUpdates = partition.map { row =>
      // 按实际_id、数组字段的类型调整对应getAs的泛型
      val docId = row.getAs[Int]("_id")
      val appendElements = row.getAs[Seq[String]]("field1")

      val filter = new Document("_id", docId)
      // 允许重复元素用pushEach，需要去重用addToSetEach
      val updateOp = pushEach("field1", appendElements: _*)
      // upsert设为true表示如果对应_id不存在则新建文档，不需要可设为false
      new com.mongodb.client.model.UpdateOneModel[Document](filter, updateOp, new UpdateOptions().upsert(true))
    }.toList

    if (batchUpdates.nonEmpty) {
      collection.bulkWrite(batchUpdates)
    }
  }
}

• 该方案的写入性能略低于原生DataFrame.write接口，适合大部分业务场景，超大数据量写入时可拆分批次降低MongoDB压力
• 不需要数组元素去重时优先选择$push算子，性能优于$addToSet

内容的提问来源于stack exchange，提问作者Puvi