使用mongo-spark connector读取MongoDB时部分字段缺失求助

我之前确实碰到过类似的情况，大概率是Schema推断或者Connector配置的问题，给你几个可行的解决方案：

• 强制指定自定义Schema
  Spark读取MongoDB时默认会采样部分文档推断Schema，如果采样的文档里刚好没有user_email这类字段，就会导致该列被排除。手动定义Schema可以彻底解决这个问题，不管文档是否包含该字段，都会保留对应的列（缺失值显示为null）。示例代码如下：

from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType

# 按照你的文档结构定义Schema
custom_schema = StructType([
    StructField("_id", StringType(), nullable=True),
    StructField("_updated", TimestampType(), nullable=True),
    StructField("total", DoubleType(), nullable=True),
    StructField("subtotal", DoubleType(), nullable=True),
    StructField("user_email", StringType(), nullable=True)
])

# 读取时指定自定义Schema
orders = spark.read.format("com.mongodb.spark.sql.DefaultSource") \
    .option("uri", "mongodb://127.0.0.1/company.orders") \
    .schema(custom_schema) \
    .load()

• 调整Schema推断的采样数量
  如果不想手动写Schema，可以调大Connector的采样文档数量，让它更大概率采样到包含目标字段的文档。默认采样数是1000，你可以通过spark.mongodb.input.schema.inference.sampleSize参数调整：

orders = spark.read.format("com.mongodb.spark.sql.DefaultSource") \
    .option("uri", "mongodb://127.0.0.1/company.orders") \
    .option("spark.mongodb.input.schema.inference.sampleSize", "5000")  # 采样5000个文档
    .load()

• 检查字段名的一致性
  确认你查询的字段名和MongoDB文档里的完全一致——Spark是区分大小写的，比如User_Email和user_email会被当成两个不同的字段，别因为大小写拼写错误导致读不到数据。

• 确认Connector版本兼容性
  版本不兼容也可能导致这类异常，要确保你的mongo-spark connector版本和Spark、MongoDB版本匹配：比如Spark 3.x需要搭配Connector 10.x及以上版本，MongoDB建议用4.2+版本。如果版本不匹配，升级到对应兼容的版本试试。

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