PySpark读取Google Cloud Bucket中JSON文件时跳过空文件的方法

问题核心分析 #

仅含[]的空JSON文件会干扰Spark的Schema自动推断；若指定的Schema与有效JSON结构不匹配，还会导致所有数据解析为空。直接读取所有文件时，空文件不会生成数据行，但Schema错误会让有效文件的数据也无法正常解析。

解决方案 #

方案一：先过滤空文件，再读取有效数据 #

先遍历所有JSON文件，筛选出内容非空列表的文件，再用匹配的Schema读取，避免空文件干扰。

代码示例

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

# 初始化SparkSession
spark = SparkSession.builder.appName("LoadValidGCSJSON").getOrCreate()

# 1. 获取GCS桶下所有JSON文件的路径（递归遍历子目录）
all_json_paths = spark.sparkContext.wholeTextFiles("gs://your-bucket/**/*.json").keys()

# 2. 过滤掉内容为[]的文件
def is_non_empty_json(path):
    # 读取文件内容并去除首尾空白字符
    content = spark.read.text(path).first()[0].strip()
    return content != "[]"

valid_paths = all_json_paths.filter(is_non_empty_json)
# 转换为本地列表供Spark读取
valid_files = [path for path in valid_paths.collect()]

# 3. 定义与有效JSON匹配的Schema（替换为你的实际结构）
custom_schema = StructType([
    StructField("user_id", IntegerType(), nullable=True),
    StructField("user_name", StringType(), nullable=True),
    StructField("email", StringType(), nullable=True)
    # 添加其他字段...
])

# 4. 读取有效文件生成DataFrame
final_df = spark.read.schema(custom_schema).json(valid_files)

# 验证结果
final_df.show()

性能优化：先按文件大小过滤

如果空文件数量极大，先通过文件大小过滤（[]一般仅占2字节，加换行最多3字节），再验证内容，减少IO开销：

from py4j.java_gateway import java_import

java_import(spark._jvm, "org.apache.hadoop.fs.Path")
fs = spark._jvm.org.apache.hadoop.fs.FileSystem.get(spark._jsc.hadoopConfiguration())

def is_valid_file(path):
    hadoop_path = spark._jvm.Path(path)
    # 跳过小于3字节的文件（排除[]及带换行的情况）
    if fs.getFileStatus(hadoop_path).getLen() < 3:
        return False
    # 二次验证内容是否为[]
    content = spark.read.text(path).first()[0].strip()
    return content != "[]"

valid_paths = all_json_paths.filter(is_valid_file)

方案二：先获取有效Schema，再读取所有文件 #

如果能找到一个已知的有效JSON文件，先读取它获取正确的Schema，再用该Schema读取所有文件——空文件不会生成数据行，最终DF自动保留有效数据。

代码示例

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("LoadValidJSONWithSchema").getOrCreate()

# 1. 读取一个已知的有效JSON文件，获取正确Schema
sample_valid_file = "gs://your-bucket/path-to-valid-sample.json"
sample_df = spark.read.json(sample_valid_file)
valid_schema = sample_df.schema

# 2. 递归读取所有JSON文件，使用获取到的Schema
final_df = spark.read.schema(valid_schema)\
    .option("recursiveFileLookup", "true")\
    .json("gs://your-bucket/**/*.json")

# 验证结果
final_df.show()

关键注意事项 #

• Schema匹配：自定义Schema必须与有效JSON的结构完全一致，包括字段名、数据类型、嵌套层级，否则有效数据会被解析为空。
• 性能考量：如果GCS上文件数量极多，方案一的过滤步骤会增加额外IO，此时优先用方案二，确保Schema正确即可。

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