PySpark读取HDFS中70GB BSON文件写入Elastic时遇序列化错误求助

从你遇到的情况来看——处理20GB数据完全正常，但读取全量70GB时就抛出Expected size to be 1179749, not 5的序列化错误——这个问题大概率和文件完整性或者Spark读取BSON的分片逻辑有关，下面是具体的排查和解决步骤：

1. 先排查HDFS上的BSON文件是否损坏 #

这个错误的核心是读取到的BSON文档实际长度和预期不匹配，首先要确认是不是某个文件/块出了问题：

• 定位出错文件：翻找Spark日志，找到报错对应的具体HDFS文件路径，然后用hdfs dfs -cat <目标文件路径> | head -n 20查看文件开头，判断是不是合法的BSON格式（正常BSON是二进制格式，乱码是正常的，但如果开头是明显的文本截断或者奇怪字符，大概率是损坏了）。
• 校验文件完整性：如果这个BSON文件是从本地或其他存储上传到HDFS的，对比本地源文件和HDFS文件的MD5/SHA哈希值，确认传输过程中没有丢包或截断。
• 检查HDFS块状态：执行hdfs fsck <文件路径>，查看文件的所有块是否都处于健康状态，有没有缺失、损坏的块。

2. 调整Spark读取BSON的分片策略 #

Spark默认的分片机制可能会把一个完整的BSON文档拆到两个分片中，导致解析失败，尤其是大文件场景：

• 手动设置分片大小：读取前修改spark.sql.files.maxPartitionBytes参数，比如设置为2g（根据你的单BSON文档大小调整，确保每个分片能包含完整的文档），示例代码：

  spark.conf.set("spark.sql.files.maxPartitionBytes", "2g")
  val bsonRdd = spark.read.format("com.mongodb.spark.sql.DefaultSource").load("hdfs://your/bson/file/path")
  

• 改用专用BSON读取器：如果你之前用的是通用文件读取方式，换成MongoDB Spark Connector的专用读取逻辑，它对BSON的结构解析更友好。

3. 容错处理：跳过损坏的记录 #

如果确实存在少量损坏的文档，且可以接受少量数据丢失，可以让Spark跳过解析失败的记录：

val bsonDF = spark.read.option("mode", "DROPMALFORMED")
  .format("com.mongodb.spark.sql.DefaultSource")
  .load("hdfs://your/bson/file/path")

也可以用PERMISSIVE模式，把解析失败的记录放到_corrupt_record字段里，后续再单独处理这些脏数据。

4. 回溯BSON文件的生成逻辑 #

如果这个BSON文件是你自己的程序生成的，检查写入过程：

• 确认写入时没有中途中断（比如程序崩溃、网络断开）导致文件截断。
• 检查生成BSON的代码，有没有出现字段类型错误、非法字符，导致文档长度计算异常。

建议先拿几个小文件做验证，确认每个子文件都能正常读取后，再逐步扩大到全量数据，这样更容易定位问题。

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