使用Spark Mongo Connector读取Mongo时如何跳过数据类型不匹配行

针对你使用spark-mongo-connector 10.1.1、Delta 2.3、EMR 3.3读取MongoDB亿级数据时遇到的类型不匹配问题（如日期字段混入字符串、整数字段混入字符串），由于Connector不支持DROPMALFORMED，可以采用以下几种实用方案：

方案一：Mongo端聚合管道提前过滤/转换（推荐大数据量场景） #

利用MongoDB的聚合能力，在数据读取到Spark之前就完成类型校验、转换和过滤，减少Spark端的数据处理压力，适合亿级数据场景。

场景1：直接过滤类型不匹配的行 #

如果只需要保留原生类型符合要求的文档，可通过$match阶段过滤：

val pipeline = """[
  {
    $match: {
      $and: [
        { field1: { $type: "date" } }, // 仅保留field1为Date类型的文档
        { field2: { $type: "int" } }   // 仅保留field2为Int类型的文档
      ]
    }
  }
]"""

val validDF = spark.read
  .format("mongodb")
  .option("uri", "mongodb://your-mongo-uri")
  .option("database", "your-db")
  .option("collection", "your-coll")
  .option("pipeline", pipeline)
  .schema(yourTargetSchema) // 直接使用预定义Schema
  .load()

场景2：尝试转换兼容类型后再过滤 #

如果部分字符串格式的日期/数值可以转换为目标类型（如"2023-10-01"转为Date），可在聚合管道中先尝试转换，再过滤转换失败的行：

val pipelineWithConversion = """[
  {
    $addFields: {
      converted_field1: {
        $cond: [
          { $eq: [{ $type: "$field1" }, "string"] },
          { $dateFromString: { dateString: "$field1", format: "%Y-%m-%d %H:%M:%S" } }, // 匹配你的字符串日期格式
          { $cond: [ { $eq: [{ $type: "$field1" }, "date"] }, "$field1", null ] }
        ]
      },
      converted_field2: {
        $cond: [
          { $eq: [{ $type: "$field2" }, "string"] },
          { $toInt: "$field2" },
          { $cond: [ { $eq: [{ $type: "$field2" }, "int"] }, "$field2", null ] }
        ]
      }
    }
  },
  {
    $match: {
      converted_field1: { $ne: null },
      converted_field2: { $ne: null }
    }
  },
  {
    $project: {
      field1: "$converted_field1",
      field2: "$converted_field2",
      other_field: 1 // 保留其他需要的字段
    }
  }
]"""

val processedDF = spark.read
  .format("mongodb")
  .option("uri", "mongodb://your-mongo-uri")
  .option("database", "your-db")
  .option("collection", "your-coll")
  .option("pipeline", pipelineWithConversion)
  .schema(yourTargetSchema)
  .load()

方案二：Spark端宽松读取后校验转换 #

先以宽松模式读取所有数据，再通过Spark内置函数校验转换字段，过滤无效行。

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

// 预定义目标Schema
val targetSchema = StructType(Seq(
  StructField("field1", TimestampType, nullable = false),
  StructField("field2", IntegerType, nullable = false),
  // 其他字段...
))

// 宽松读取所有数据
val rawDF = spark.read
  .format("mongodb")
  .option("uri", "mongodb://your-mongo-uri")
  .option("database", "your-db")
  .option("collection", "your-coll")
  .load()

// 处理字段并过滤无效行
val validDF = rawDF
  // 处理field1：兼容Date/String类型，转换失败返回null
  .withColumn("field1", 
    when(col("field1").cast(TimestampType).isNotNull, col("field1").cast(TimestampType))
    .when(to_timestamp(col("field1"), "yyyy-MM-dd HH:mm:ss").isNotNull, to_timestamp(col("field1"), "yyyy-MM-dd HH:mm:ss"))
    .otherwise(null)
  )
  // 处理field2：兼容Int/String类型，转换失败返回null
  .withColumn("field2", 
    when(col("field2").cast(IntegerType).isNotNull, col("field2").cast(IntegerType))
    .otherwise(null)
  )
  // 过滤所有必填字段转换成功的行
  .filter(col("field1").isNotNull && col("field2").isNotNull)
  // 匹配目标Schema
  .withSchema(targetSchema)

方案三：自定义UDF处理复杂校验逻辑 #

如果内置函数无法满足复杂的类型校验规则（如特殊格式的日期字符串），可编写自定义UDF处理转换，失败返回null后过滤。

import org.apache.spark.sql.functions.udf
import java.sql.Timestamp
import java.text.SimpleDateFormat

// 自定义日期转换UDF
val parseDateUDF = udf((value: Any) => {
  value match {
    case ts: Timestamp => ts
    case d: java.util.Date => new Timestamp(d.getTime)
    case s: String =>
      try {
        val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
        new Timestamp(sdf.parse(s).getTime)
      } catch {
        case _: Exception => null
      }
    case _ => null
  }
})

// 自定义整数转换UDF
val parseIntUDF = udf((value: Any) => {
  value match {
    case i: Int => i
    case s: String =>
      try {
        s.toInt
      } catch {
        case _: Exception => null
      }
    case _ => null
  }
})

// 处理数据
val validDF = rawDF
  .withColumn("field1", parseDateUDF(col("field1")))
  .withColumn("field2", parseIntUDF(col("field2")))
  .filter(col("field1").isNotNull && col("field2").isNotNull)
  .withSchema(targetSchema)

性能优化建议 #

• 优先选择方案一，Mongo端过滤可大幅减少Spark读取的数据量，提升亿级数据处理效率
• 读取时配置合理的分区策略：option("partitioner", "MongoPaginateBySizePartitioner") + option("partitionSizeMB", "64")（根据集群资源调整）
• 尽量使用Spark内置函数替代UDF，减少序列化开销
• 处理后的有效数据可写入Delta表，利用Delta的优化特性提升后续查询性能

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