嘿，我来帮你搞定这个Spark读取Elasticsearch时的Schema推断问题！

你遇到的核心问题是：Spark自动推断ES Schema时，会抽样部分文档来识别字段类型。如果你的嵌套JSON字段存在类型不一致（比如同一个字段在某些文档里是数字，另一些是字符串），或者字段值本身格式违规（比如看似数字的字符串里混了非数字字符），就会触发NumberFormatException，直接导致DataFrame加载失败。

1. 手动指定Schema（最稳妥的方案） #

既然自动推断容易踩坑，不如直接手动定义精准的Schema，强制指定每个字段（包括嵌套字段）的类型，彻底绕开自动推断的问题。

给你贴个适配嵌套结构的代码示例：

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

// 自定义包含嵌套结构的Schema，把出问题的字段设为正确类型
val customEsSchema = StructType(Array(
  StructField("doc_id", StringType, nullable = true),
  StructField("user_info", StructType(Array(
    StructField("age", IntegerType, nullable = true), // 明确指定整数类型
    StructField("custom_tag", StringType, nullable = true) // 如果这个字段曾触发异常，就指定为字符串
  )), nullable = true)
))

// 读取ES时绑定自定义Schema
val esDf = spark.read
  .format("org.elasticsearch.spark.sql")
  .schema(customEsSchema)
  .option("es.nodes", "你的ES主机地址")
  .option("es.port", "9200")
  .load("你的索引名/_doc")

2. 调整Spark的Schema推断配置 #

你可以通过修改配置让Spark的推断逻辑更“宽容”：

• 提高抽样文档数：默认Spark只抽样少量文档，可能刚好抽到了有问题的样本。可以设置es.read.schema.sample.size参数（比如设为10000），让Spark抽样更多文档来推断Schema，降低抽到异常样本的概率。
• 开启空值转换：设置es.read.field.convert.empty.to.null为true，把空值或格式异常的字段转为null，避免解析报错。

3. 修复Elasticsearch端的数据/映射 #

如果数据本身存在格式混乱，最根本的解决办法是从ES端入手：

• 检查并修正ES索引的Mapping：确保嵌套字段的类型是统一的（比如如果字段应该存字符串，就把Mapping设为keyword或text，不要让ES自动推断类型）。
• 批量修正异常文档：把嵌套字段里格式违规的值（比如字符串形式的数字）转换成统一格式，或者直接清理掉无效数据。

4. 启用容错读取模式 #

如果你能接受丢失少量异常文档或字段为null，可以开启Spark的容错读取模式：

• permissive模式：把解析失败的字段设为null，保留其他正常字段
• dropMalformed模式：直接丢弃整个解析失败的文档

代码示例：

val esDf = spark.read
  .format("org.elasticsearch.spark.sql")
  .option("es.nodes", "你的ES主机地址")
  .option("es.port", "9200")
  .option("mode", "permissive") // 替换成"dropMalformed"即可切换模式
  .load("你的索引名/_doc")

内容的提问来源于stack exchange，提问作者Jasjyot Singh Jaswal