在Databricks中用PySpark加载S3千万JSON文件不全的问题排查

以下是针对你遇到的Spark加载S3 JSON文件不完整问题的常见排查点和解决办法：

1. 检查文件路径匹配规则 #

• Spark的json()方法默认会递归遍历子目录，但如果文件分布在更深层级，或者路径未覆盖所有目标文件，可尝试显式使用递归通配符：

  df = spark.read.schema(schema).json("s3://data_bucket/json_files/**")
  

• 排查是否存在隐藏文件或命名异常的文件：比如以.、_开头的文件，或者后缀非.json的文件（虽然Spark不强制后缀，但部分场景下会被过滤）。可以用HDFS API列举所有文件统计数量：

  fs = spark.sparkContext._jvm.org.apache.hadoop.fs.FileSystem.get(spark.sparkContext._jsc.hadoopConfiguration())
  file_list = fs.listStatus(spark._jvm.org.apache.hadoop.fs.Path("s3://data_bucket/json_files/"))
  print(f"总文件数：{len(file_list)}")
  # 过滤异常文件名
  odd_files = [f.getPath().toString() for f in file_list if any(x in f.getPath().getName() for x in [".", "_", "tmp"])]
  

2. 验证JSON文件格式有效性 #

• 大部分加载失败的情况是因为文件不是JSON Lines格式（每行一个独立JSON对象），而是多行嵌套的JSON结构。Spark默认按行解析，这类文件会被判定为格式错误跳过，可开启multiLine选项：

  df = spark.read.schema(schema).option("multiLine", "true").json("s3://data_bucket/json_files/")
  

• 检查空文件或损坏文件：空文件会被Spark跳过，可筛选出大小为0的文件：

  file_sizes = [(f.getPath().toString(), f.getLen()) for f in file_list]
  empty_files = [path for path, size in file_sizes if size == 0]
  print(f"空文件数量：{len(empty_files)}")
  

• 查看Spark日志中的解析错误：在Databricks集群的Spark UI或日志面板中搜索corrupt_record、JSON parsing error，定位具体损坏的文件。

3. 确认Schema兼容性 #

• 自定义Schema与部分文件结构不匹配时，Spark可能会跳过整个文件或解析为空记录。可先小范围测试自动推断Schema，对比与你定义的Schema差异：

  # 取部分目录测试
  sample_df = spark.read.json("s3://data_bucket/json_files/subdir_1/")
  print("推断的Schema：")
  sample_df.printSchema()
  

• 开启PERMISSIVE模式（默认开启）后，可通过_corrupt_record字段查看解析失败的记录，定位问题文件：

  df = spark.read.schema(schema).option("mode", "PERMISSIVE").json("s3://data_bucket/json_files/")
  df.filter(df._corrupt_record.isNotNull()).select("_corrupt_record").show(truncate=False)
  

4. 排查S3权限与一致性问题 #

• 确认Databricks集群的IAM角色拥有所有目标文件的s3:GetObject权限：可在集群节点上用AWS CLI测试访问未加载的文件：

  aws s3 cp s3://data_bucket/json_files/xxx.json /tmp/test.json
  

• S3最终一致性可能导致近期上传的文件未被列举：如果文件是刚上传的，等待1-2分钟后重新加载，或者用S3控制台确认文件存在。

5. 调整Spark文件列举配置 #

• 清除Spark的文件列表缓存：

  spark.catalog.clearCache()
  

• 增大S3文件列举的最大项数，避免因列举限制遗漏文件：

  spark.conf.set("fs.s3a.list.max-items", "1000000")
  

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