如何通过Spark基于时间戳条件更新MongoDB文档

需求场景 #

现有MongoDB文档结构：

_id:63d7cc8c30f3fddc1d4a266b
name: "nameA"
address: "Paris"
timestamp: "2023-04-17 16:43:00"

Spark DataFrame数据：

+-----+------+-------------------+
|name |address|timestamp          |
+-----+------+-------------------+
|nameA|Berlin|2023-04-15 12:23:00|
|nameA|London|2023-04-18 09:54:00|
+-----+------+-------------------+

要求：按name字段匹配MongoDB文档，仅当DataFrame中的timestamp大于MongoDB现有文档的timestamp时，才更新address和timestamp字段。最终MongoDB文档应更新为：

_id:63d7cc8c30f3fddc1d4a266b
name: "nameA"
address: "London"
timestamp: "2023-04-18 09:54:00"

实现步骤 #

直接使用SaveMode.Overwrite会覆盖目标集合或匹配文档，无法实现条件过滤。需结合Spark数据关联+MongoDB连接器的更新操作来实现：

1. 统一数据类型（关键） #

先将MongoDB和DataFrame中的timestamp字段转换为Timestamp类型，避免字符串比较的潜在逻辑错误：

import org.apache.spark.sql.types.TimestampType
import org.apache.spark.sql.functions.to_timestamp

// 处理DataFrame的timestamp字段
val dfProcessed = df.withColumn("timestamp", to_timestamp($"timestamp", "yyyy-MM-dd HH:mm:ss"))

// 读取MongoDB现有数据并处理timestamp
val mongoDf = spark.read
  .format("mongodb")
  .option(MongoConfig.CONNECTION_STRING_CONFIG, ConfigMongoDb.getUri(configMongoDb))
  .option(MongoConfig.DATABASE_NAME_CONFIG, configMongoDb.database)
  .option(MongoConfig.COLLECTION_NAME_CONFIG, collection)
  .load()
  .withColumn("mongo_timestamp", to_timestamp($"timestamp", "yyyy-MM-dd HH:mm:ss"))

2. 关联过滤符合条件的更新记录 #

将处理后的DataFrame与MongoDB数据按name关联，筛选出DataFrame中时间戳更大的记录：

import org.apache.spark.sql.functions.col

val updateRecords = dfProcessed.join(
  mongoDf.select("name", "mongo_timestamp"),
  Seq("name"),
  "inner"
).where(col("timestamp") > col("mongo_timestamp"))
// 只保留需要更新的字段
.select("name", "address", "timestamp")

此时updateRecords中只会保留nameA的London那条记录，因为它的时间戳大于MongoDB中的现有值。

3. 执行条件更新到MongoDB #

使用Spark-MongoDB连接器的更新模式，指定匹配条件和更新逻辑：

updateRecords.write
  .format("mongodb")
  .option(MongoConfig.CONNECTION_STRING_CONFIG, ConfigMongoDb.getUri(configMongoDb))
  .option(MongoConfig.DATABASE_NAME_CONFIG, configMongoDb.database)
  .option(MongoConfig.COLLECTION_NAME_CONFIG, collection)
  // 指定操作类型为更新
  .option("operationType", "update")
  // 按name字段匹配目标文档
  .option("updateFilter", "{ 'name': ?name }")
  // 使用$set操作符仅更新指定字段，避免覆盖_id等其他字段
  .option("updateDocument", "{ '$set': { 'address': ?address, 'timestamp': ?timestamp } }")
  // 确保每个匹配条件只更新一条文档（适合同name唯一的场景）
  .option("updateOne", "true")
  // 若无匹配文档则不插入（按需调整为true可插入新文档）
  .option("upsert", "false")
  .save()

关键参数说明 #

• operationType: update：告知连接器执行更新操作而非插入/覆盖
• updateFilter：指定MongoDB的匹配规则，?name会自动绑定DataFrame中的name字段值
• updateDocument：使用MongoDB的$set操作符，仅更新指定字段，保留原文档其他内容
• updateOne: true：避免同name多条记录时重复更新
• upsert: false：无匹配文档时不插入新数据，可根据业务需求调整

验证结果 #

执行后，MongoDB中name: "nameA"的文档会被更新为目标结构，Berlin那条时间戳更早的记录不会触发更新。

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