Spark加载MongoDB DataFrame时报错：Cannot cast ARRAY into a StructType

这个问题本质是MongoDB动态Schema特性和Spark强类型Schema的冲突——目标字段在MongoDB文档里同时存在对象（StructType）、**数组（ArrayType）**和null值，Spark自动推断Schema时无法统一类型，导致展示或处理数据时抛出类型转换错误。

以下是几种可行的解决思路：

1. 自定义Schema并提前过滤/转换混合类型 #

先明确该字段的业务预期类型（是Struct还是Array），自定义Schema强制指定后，对不符合类型的数据进行处理：

• 如果预期是StructType：过滤掉数组类型的数据，或者把数组的第一个元素转为Struct
• 如果预期是ArrayType：把单个Struct转为单元素数组

示例代码（Scala）：

import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._

// 定义目标Struct类型的Schema
val targetStructSchema = StructType(Seq(
  StructField("fieldA", StringType),
  StructField("fieldB", IntegerType)
))

val customSchema = StructType(Seq(
  StructField("docId", StringType),
  StructField("targetField", targetStructSchema, nullable = true)
))

val df = spark.read
  .format("mongodb")
  .option("uri", "mongodb://localhost:27017/your_db.your_collection")
  .schema(customSchema)
  .load()
  // 方案1：过滤掉数组类型的记录
  .filter(expr("NOT is_array(targetField)"))
  // 方案2：将数组的第一个元素转为Struct保留
  // .withColumn("targetField", when(is_array(col("targetField")), col("targetField").getItem(0)).otherwise(col("targetField")))

2. 利用MongoDB连接器的混合类型处理选项 #

新版本的Spark MongoDB Connector提供了mixedTypesHandling配置，可强制统一混合类型的字段：

• 设置为struct：将数组类型转为Struct（取第一个元素）
• 设置为array：将Struct类型转为单元素数组

示例代码：

val df = spark.read
  .format("mongodb")
  .option("uri", "mongodb://localhost:27017/your_db.your_collection")
  // 根据业务需求选struct或array
  .option("spark.mongodb.input.mixedTypesHandling", "struct")
  .load()

3. 先以字符串读取，再动态解析类型 #

如果无法提前确定类型，可以先将该字段以StringType读取，之后再根据内容动态转换为Struct或Array：

import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._

val tempSchema = StructType(Seq(
  StructField("docId", StringType),
  StructField("targetField", StringType, nullable = true)
))

val targetStructSchema = StructType(Seq(
  StructField("fieldA", StringType),
  StructField("fieldB", IntegerType)
))

val df = spark.read
  .format("mongodb")
  .option("uri", "mongodb://localhost:27017/your_db.your_collection")
  .schema(tempSchema)
  .load()
  .withColumn("targetField", when(col("targetField").isNull, lit(null))
    .otherwise(
      // 判断是否为Struct格式，是则解析为Struct，否则解析为Array
      when(expr("json_tuple(targetField, 'fieldA') is not null"), 
           from_json(col("targetField"), targetStructSchema))
      .otherwise(from_json(col("targetField"), ArrayType(targetStructSchema)))
    )
  )

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