MongoDB到Databricks流写入DeltaTable报错求助

错误原因分析 #

错误提示Change stream must be followed by a match and then a project stage来自Azure Cosmos DB for MongoDB（你的服务器是CosmosDB的Mongo兼容版），核心问题是你同时启用了spark.mongodb.change.stream.publish.full.document.only配置和自定义pipeline，导致MongoDB接收到的变更流管道结构不符合校验要求。

publish.full.document.only选项会自动向变更流管道插入一个$project阶段来提取fullDocument字段，但你自定义的pipeline中又重复添加了$match和$project，这会让实际执行的管道顺序混乱，触发CosmosDB的规则校验错误。

解决方案 #

1. 移除冲突配置项 #

删除spark.mongodb.change.stream.publish.full.document.only=true配置，因为你的自定义pipeline已经包含了提取fullDocument的逻辑，重复配置会导致管道结构异常。

2. 修正Pipeline格式与逻辑 #

确保pipeline使用标准双引号的合法JSON格式，同时严格保持$match在前、$project在后的执行顺序。

3. 修改后的完整代码 #

def doThis(batch_data, batch_id):
    batch_data.write.mode("append").format("delta").saveAsTable("demo.practice_schema.demo_mongodb")

dataStreamWriter = (spark.readStream
                        .format("mongodb")
                        .option("spark.mongodb.connection.uri", connection_string)
                        .option('spark.mongodb.database', database_name)
                        .option('spark.mongodb.collection', current_collection_name)
                        # 移除冲突的publish.full.document.only配置
                        .option("pipeline", '[{"$match":{"operationType":{"$in": ["insert", "update", "replace"]}}},{"$project":{"fullDocument":1, "_id":0}}]')
                        .load()
                        .select("fullDocument.*")  # 展开嵌套的fullDocument字段，避免Delta表仅存嵌套对象
                        .writeStream
                        .foreachBatch(doThis)
                        .start())

额外注意事项 #

• 加入.select("fullDocument.*")是为了展开嵌套的fullDocument字段，让Delta表直接存储业务数据字段，而非嵌套对象，方便后续查询分析。
• 如果你需要同步多个结构差异较大的集合，建议为每个集合单独创建Delta表，或者在写入前通过Spark的schema合并、字段映射等操作统一数据结构，避免出现schema不一致的写入错误。

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