Spark读取MongoDB报错：无法将ARRAY转为StructType求解决方案

错误信息 #

com.mongodb.spark.exceptions.MongoTypeConversionException: Cannot cast ARRAY into a StructType(StructField(userid,StringType,true),StructField(username,StringType,true),StructField(zohocrm_id,StringType,true)) (value: BsonArray{values=[]})

问题场景 #

使用Spark加载MongoDB全量数据时触发上述错误，但在pipeline中加入$match过滤后能正常运行。需要不依赖$match实现全量数据加载。

当前代码 #

spark = SparkSession \
        .builder \
        .appName("mongo4XC") \
        .config("spark.mongodb.input.uri",dbURLinfo) \
        .config("spark.mongodb.output.uri",dbURLinfo) \
        .config("spark.executor.heartbeatInterval", "180s") \
        .config("spark.network.timeout", "300s") \
        .config('spark.jars.packages', 'org.mongodb.spark:mongo-spark-connector_2.12:3.0.2') \
        .config("spark.sql.debug.maxToStringFields", 1000) \
        .config("spark.mongodb.input.sampleSize", 500000) \
        .getOrCreate()

df = spark.read.format("com.mongodb.spark.sql.DefaultSource") \
        .option("database", db_database) \
        .option("collection", collection) \
        .option("pipeline", pipeline) \
        .load()

问题根因 #

MongoDB集合中部分文档的同名字段类型不统一：大部分文档里该字段是符合StructType的嵌套结构，但少量文档中这个字段被错误存储为空数组[]。使用$match时恰好过滤掉了这些异常文档，因此能正常运行；加载全量数据时，Spark Connector通过采样推断Schema后，遇到数组类型的同名字段就会触发类型转换冲突。

解决方案 #

方案1：强制指定Schema，关闭自动推断

手动定义目标Schema，让Connector忽略采样结果，确保类型匹配。若需要兼容数组类型的异常字段，可在读取后进行转换。

from pyspark.sql.types import StructType, StructField, StringType
from pyspark.sql.functions import col, when, struct

# 定义目标Schema
custom_schema = StructType([
    StructField("userid", StringType(), nullable=True),
    StructField("username", StringType(), nullable=True),
    StructField("zohocrm_id", StringType(), nullable=True),
    # 补充其他字段
])

# 读取时指定Schema并关闭自动推断
df = spark.read.format("com.mongodb.spark.sql.DefaultSource") \
        .option("database", db_database) \
        .option("collection", collection) \
        .option("spark.mongodb.input.schemaInference", "false") \
        .schema(custom_schema) \
        .load()

# 可选：将数组类型的异常字段转为默认空结构
df_clean = df.withColumn("target_field", 
    when(col("target_field").isArray(), struct(
        None.cast(StringType()).alias("userid"),
        None.cast(StringType()).alias("username"),
        None.cast(StringType()).alias("zohocrm_id")
    ))
    .otherwise(col("target_field"))
)

方案2：在MongoDB Pipeline中预处理异常数据

无需过滤文档，用$addFields统一字段类型，将数组转为符合结构的默认值后再加载。

# 假设异常字段名为user_info
pipeline = [
    {
        "$addFields": {
            "user_info": {
                "$cond": {
                    "if": {"$isArray": "$user_info"},
                    "then": {"userid": None, "username": None, "zohocrm_id": None},
                    "else": "$user_info"
                }
            }
        }
    }
]

df = spark.read.format("com.mongodb.spark.sql.DefaultSource") \
        .option("database", db_database) \
        .option("collection", collection) \
        .option("pipeline", pipeline) \
        .load()

方案3：调整采样策略（临时应急）

当前设置sampleSize=500000，若采样覆盖到异常文档会导致Schema推断错误：

• 减小sampleSize避开异常文档（治标不治本，不推荐）
• 设置spark.mongodb.input.schemaInference.sampleSize=0，强制扫描全量文档推断Schema（仅适合小集合，性能较差）

内容的提问来源于stack exchange，提问作者Mai Nhật Nam