如何将MongoDB复杂数据加载到PySpark DataFrame并处理Schema与扁平化？

AI 小北

2026-5-14

Hey there! Let's tackle your Spark + MongoDB issues step by step—this is a common pain point when dealing with nested/complex schemas, so let's break it down.

1. Is there a better loading method than reading complex fields as strings?

Absolutely. The error cannot cast DOCUMENT in to ArrayType tells me the root issue: your MongoDB collection has inconsistent data types for the same field (e.g., some documents store cd as an array, others as a single embedded document). This confuses Spark's automatic schema inference. Here are your best fixes, ordered by efficiency:

Option 1: Explicitly define your target schema

Since you already know the exact MongoDB schema, pass it directly to the Spark reader. This forces the MongoDB Connector to parse data strictly against your schema, avoiding inference errors.

Example code (Scala):

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

// Define your full schema exactly as you provided
val targetSchema = StructType(List(
  StructField("_id", StructType(List(StructField("oid", StringType, true))), true),
  StructField("_meta", StructType(List(
    StructField("ind", StringType, true),
    StructField("updt", StringType, true),
    StructField("clid", StringType, true),
    StructField("ddr", StringType, true)
  )), true),
  StructField("cd", ArrayType(StructType(List(
    StructField("cn", StringType, true),
    StructField("rs", StringType, true),
    StructField("ck", ArrayType(StringType, true), true)
  )), true), true),
  StructField("date", TimestampType, true),
  StructField("mid", StringType, true),
  StructField("lm", ArrayType(StructType(List(
    StructField("code", StringType, true),
    StructField("ed", StringType, true),
    StructField("lid", StringType, true),
    StructField("lk", StringType, true),
    StructField("status", StringType, true),
    StructField("type", StringType, true),
    StructField("pn", StringType, true),
    StructField("pv", StringType, true),
    StructField("nk", StringType, true),
    StructField("id", StringType, true),
    StructField("dt", StringType, true),
    StructField("mi", StringType, true)
  )), true), true),
  StructField("expo", ArrayType(StructType(List(
    StructField("ek", StringType, true),
    StructField("ev", ArrayType(StructType(List(
      StructField("en", StringType, true),
      StructField("ev", StringType, true)
    )), true), true)
  )), true), true),
  StructField("pim", StringType, true),
  StructField("pis", StringType, true),
  StructField("piv", StringType, true),
  StructField("pvw", StringType, true)
))

// Load data with the explicit schema
val df = sqlContext.read
  .format("com.mongodb.spark.sql.DefaultSource")
  .option("uri", uri)
  .option("database", db)
  .option("collection", ct)
  .schema(targetSchema) // Critical: enforce your schema here
  .load()

Option 2: Adjust schema inference sampling

If inference fails because Spark's default sample size is too small (it misses documents with rare type variations), increase the sample size:

val df = sqlContext.read
  .format("com.mongodb.spark.sql.DefaultSource")
  .option("uri", uri)
  .option("database", db)
  .option("collection", ct)
  .option("spark.mongodb.input.schema.inference.sampleSize", "10000") // Sample 10k docs
  .load()

Option 3: Clean inconsistent data in MongoDB

If some documents have non-standard types (e.g., a single cd object instead of an array), fix this at the source first. For example, use a MongoDB update to wrap single objects in arrays:

db.yourCollection.updateMany(
  { cd: { $type: "object" } },
  { $set: { cd: ["$cd"] } }
)

All these options are far better than reading as strings—they give you a structured DataFrame immediately, no extra parsing needed.

2. If you must load complex fields as strings, how to flatten them optimally?

If you can't fix the source data or schema inference, use Spark's JSON functions to parse and flatten the string fields. Here's a step-by-step workflow:

Step 1: Parse string fields back to complex types

Use from_json with the appropriate sub-schema to convert string-encoded structs/arrays back to their native Spark types.

Step 2: Flatten structs and explode arrays

For structs: Access sub-fields directly with . notation.
For arrays: Use explode to split arrays into individual rows, then flatten nested structs inside the array.

Example code (Scala):

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

// Parse string fields back to complex types
val parsedDF = df
  // Parse _id struct
  .withColumn("_id_struct", from_json(col("_id"), StructType(List(StructField("oid", StringType, true)))))
  // Parse _meta struct
  .withColumn("_meta_struct", from_json(col("_meta"), StructType(List(
    StructField("ind", StringType, true),
    StructField("updt", StringType, true),
    StructField("clid", StringType, true),
    StructField("ddr", StringType, true)
  ))))
  // Parse cd array
  .withColumn("cd_array", from_json(col("cd"), ArrayType(StructType(List(
    StructField("cn", StringType, true),
    StructField("rs", StringType, true),
    StructField("ck", ArrayType(StringType, true), true)
  )))))

// Flatten structs into top-level columns
val flattenedStructsDF = parsedDF
  .select(
    "_id_struct.oid" as "id_oid",
    "_meta_struct.ind" as "meta_ind",
    "_meta_struct.updt" as "meta_updt",
    "_meta_struct.clid" as "meta_clid",
    "_meta_struct.ddr" as "meta_ddr",
    "date",
    "mid",
    "cd_array",
    "pim", "pis", "piv", "pvw"
  )

// Explode the cd array and flatten its nested fields
val finalFlattenedDF = flattenedStructsDF
  .withColumn("cd", explode(col("cd_array")))
  .select(
    "*",
    "cd.cn" as "cd_cn",
    "cd.rs" as "cd_rs",
    array_join(col("cd.ck"), ",") as "cd_ck_str" // Convert ck array to comma-separated string
  )
  .drop("cd_array", "cd")

For arrays like lm or expo, repeat the explode/flatten pattern—adjust based on whether you want to keep array elements as rows or aggregate them into strings/lists.

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