Spark DataFrame写入Elasticsearch无报错但数据量缺失问题求助

1. 写入失败静默丢弃 #

Elasticsearch-Spark 连接器默认对写入失败的批次有重试机制，重试失败后默认不会抛出异常，而是直接丢弃失败批次，不会打断Spark任务的运行，这是任务无报错但数据丢失的最常见原因。

2. 重复文档ID覆盖 #

如果你配置了es.mapping.id参数指定DataFrame的某列作为ES文档主键，DataFrame中重复ID的行写入ES时会触发覆盖逻辑，相同ID只会保留最后一条，最终写入条数会小于源DataFrame总条数。

3. 字段类型不兼容 #

如果ES索引提前配置了固定Mapping，写入的字段类型和Mapping定义不匹配的行，会被ES直接丢弃且不会返回报错，比如Mapping定义为整型的字段写入了字符串值，对应行就会被静默过滤。

4. ES集群处理能力不足 #

ES节点的Bulk写入队列容量有限，你当前Spark任务的写入并发（16个executor共32核）过高，超过ES集群的处理上限时，Bulk请求会被直接拒绝，重试失败后就会丢失对应批次的数据。

5. 无效参数干扰 #

你提交的是PySpark任务，不需要指定--class org.apache.spark.examples.SparkPi参数，该参数是Java/Scala Jar任务指定主类使用的，属于无效配置，虽然当前不影响运行但可能引发后续类加载异常。

• 开启写入失败抛错配置，避免静默丢数，在写入ES的Option中新增以下配置：

.option("es.batch.write.fail.on.error", "true")
.option("es.batch.write.retry.count", "10") // 适当调高重试次数，适配ES偶尔的抖动

配置后如果存在写入错误，任务会直接抛出异常终止，你可以从Spark日志中定位具体错误原因。

• 校验重复ID：如果指定了自定义文档ID，先运行final_df.select("你指定的ID列").dropDuplicates().count()，对比去重后的条数和源DataFrame总条数是否一致，如果存在差值就是重复ID导致的覆盖。
• 校验Mapping兼容性：使用少量样本数据写入测试索引，对比写入前后的条数是否一致，如果存在差值，查看ES节点日志定位不兼容的字段类型。
• 匹配写入并发：可以适当调低Spark的executor数量，或者将es.batch.size.entries参数从默认的1000调整为500，降低单批次写入压力，适配ES集群的处理能力。
• 清理无效参数：删除spark-submit命令中的--class org.apache.spark.examples.SparkPi配置。

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