Spark Structured Streaming读取MongoDB文档转文本，新增字段未识别求替代方案

你的问题核心在于：Spark Structured Streaming启动后会固定初始Schema，micro_batch.columns仅包含流启动时识别的字段，后续MongoDB集合新增的字段不会被动态加载到columns列表中，导致to_json生成的JSON缺失新字段。

方案1：利用MongoDB Spark Connector直接获取原始JSON字符串 #

MongoDB Spark Connector支持直接将文档读取为Extended JSON格式的字符串，无需手动构造字段结构，自动保留所有新增字段。

修改读取流配置：

query = spark.readStream\
    .format("mongodb")\
    .option("spark.mongodb.connection.uri", connectionString)\
    .option("spark.mongodb.database", database)\
    .option("spark.mongodb.collection", collection)\
    .option("mode","PERMISSIVE")\
    .option("spark.mongodb.change.stream.publish.full.document.only", "true")\
    .option("spark.mongodb.input.format", "json")  # 指定读取为JSON字符串
    .option("mergeSchema", "true")\
    .load()

更新foreachBatch逻辑：

def batch(micro_batch, id):
    result_df = micro_batch\
        .withColumn("Date_Inserted", current_timestamp())\
        .withColumnRenamed("value", "json_data")\
        .withColumn("json_variant", parse_json(col("json_data")))\
        .select("Date_Inserted", "_id", "json_data", "json_variant")

    if result_df.head(1):
        result_df.write.mode("append").option("mergeSchema", "true").format("delta").saveAsTable(destination_table)

优点：无需手动处理字段，性能最优，自动兼容新增字段；注意：部分旧版本Connector可能需要用spark.mongodb.read.json替代spark.mongodb.input.format。

方案2：自定义UDF转换完整Row为JSON #

通过UDF直接将每个Row序列化为包含所有字段的JSON，绕过Spark固定Schema的限制。

先导入依赖并定义UDF：

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType
import bson.json_util

@udf(StringType())
def row_to_json(row):
    # 将Row转为字典，再用bson工具生成标准JSON
    return bson.json_util.dumps(row.asDict(recursive=True))

修改foreachBatch逻辑：

def batch(micro_batch, id):
    result_df = micro_batch\
        .withColumn("json_data", row_to_json(struct("*")))\
        .withColumn("json_variant", parse_json(col("json_data")))\
        .withColumn("Date_Inserted", current_timestamp())\
        .select("Date_Inserted", "_id", "json_data", "json_variant")

    if result_df.head(1):
        result_df.write.mode("append").option("mergeSchema", "true").format("delta").saveAsTable(destination_table)

优点：兼容嵌套结构，不依赖Connector特定配置；缺点：UDF性能略低于内置函数，需确保Databricks环境安装了pymongo库。

方案3：启用流Schema自动演进 #

通过配置让Spark动态更新Schema，结合struct("*")自动包含所有字段。

修改读取流配置，添加Schema自动推断：

query = spark.readStream\
    .format("mongodb")\
    .option("spark.mongodb.connection.uri", connectionString)\
    .option("spark.mongodb.database", database)\
    .option("spark.mongodb.collection", collection)\
    .option("mode","PERMISSIVE")\
    .option("spark.mongodb.change.stream.publish.full.document.only", "true")\
    .option("inferSchema", "true")\
    .option("mergeSchema", "true")\
    .option("spark.sql.streaming.schemaInference", "true")  # 启用Schema自动演进
    .load()

简化to_json调用：

.withColumn("json_data", to_json(struct("*")))

注意：该方式依赖Spark的Schema自动演进功能，仅适用于支持动态Schema的场景，新增字段需要等待流触发Schema更新的批次，持续运行的流可能需要重启才能识别新字段。

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