在Databricks中读取无扩展名Gzip压缩JSON文件失败的解决方法

方案1：二进制流读取后手动解析（最灵活） #

直接以二进制格式读取所有文件，手动完成gzip解压和JSON解析，完全绕过扩展名依赖：

import zlib
from pyspark.sql.functions import col, from_json, udf
from pyspark.sql.types import StringType, StructType

# 自定义UDF：解压gzip二进制数据为字符串
def gunzip_binary(binary_data):
    if binary_data is None:
        return None
    try:
        return zlib.decompress(binary_data, 16+zlib.MAX_WBITS).decode("utf-8")
    except Exception:
        return None

gunzip_udf = udf(gunzip_binary, StringType())

# 读取二进制流
binary_df = spark.readStream.format("cloudFiles") \
    .option("cloudFiles.format", "binary") \
    .option("cloudFiles.includeExistingFiles", True) \
    .option("checkpointLocation", checkpointPath) \
    .option("modifiedAfter", modifiedAfterValue) \
    .load(rawPath)

# 预定义JSON Schema（生产环境建议使用预定义Schema而非自动推断）
json_schema = StructType([
    # 替换为你的实际字段定义，例如：
    # StructField("id", IntegerType()),
    # StructField("event_time", TimestampType())
])

# 解压并解析JSON
raw_df = binary_df \
    .withColumn("json_str", gunzip_udf(col("content"))) \
    .filter(col("json_str").isNotNull()) \
    .select(from_json(col("json_str"), json_schema).alias("data")) \
    .select("data.*")

说明：

• 不依赖任何文件标识，适配所有无扩展名的Gzip压缩JSON文件
• 可在UDF中添加自定义错误逻辑，过滤损坏或不符合格式的文件
• 预定义Schema能提升流处理性能，避免自动推断带来的额外开销

方案2：强制Autoloader按指定格式解析（代码改动最小） #

通过Spark配置忽略文件扩展名检查，强制Autoloader按Gzip压缩的JSON格式读取：

# 配置Spark忽略JSON文件的扩展名校验
spark.conf.set("spark.sql.json.ignoreFileExtension", "true")

raw_df = spark.readStream.format("cloudFiles") \
    .option("cloudFiles.format", "json") \
    .option("cloudFiles.inferColumnTypes", inferColumnTypeValue) \
    .option("mergeSchema", "true") \
    .option("cloudFiles.schemaLocation", schemaPath) \
    .option("cloudFiles.allowOverwrites", "true") \
    .option("ignoreCorruptFiles", "true") \
    .option("header", headerValue) \
    .option("compression", "gzip") \
    .option("cloudFiles.includeExistingFiles", True) \
    .option("checkpointLocation", checkpointPath) \
    .option("modifiedAfter", modifiedAfterValue) \
    .option("cloudFiles.fileFilter", "*")  # 匹配目录下所有无扩展名文件
    .load(rawPath)

说明：

• spark.sql.json.ignoreFileExtension=true 让Spark跳过扩展名检查，直接按JSON格式解析内容
• compression=gzip 强制启用Gzip解压逻辑
• 保留Autoloader原生的增量读取、Schema合并等功能，代码改动最小

方案3：批量重命名历史文件（一次性处理） #

如果历史无扩展名文件较多，且后续Ingestion流程会修复扩展名问题，可通过Databricks批量添加.json.gz扩展名，之后复用原有代码：

from azure.storage.filedatalake import DataLakeServiceClient

# 替换为你的ADL Gen2存储信息
account_name = "your-storage-account"
account_key = "your-storage-key"
file_system_name = "your-container"

service_client = DataLakeServiceClient(
    account_url=f"https://{account_name}.dfs.core.windows.net",
    credential=account_key
)
file_system_client = service_client.get_file_system_client(file_system=file_system_name)

# 遍历指定目录并重命名无扩展名文件
def rename_target_files(directory_path):
    dir_client = file_system_client.get_directory_client(directory_path)
    for path in dir_client.get_paths():
        if not path.is_directory and "." not in path.name:
            new_file_name = f"{path.name}.json.gz"
            dir_client.rename_file(path.name, new_file_name)

# 示例：处理2020/11/15/23目录
rename_target_files("raw-layer/2020/11/15/23")

说明：

• 适合一次性清理历史数据，后续可继续使用原有Autoloader代码
• 需确保拥有ADL Gen2的读写权限，避免误修改其他有扩展名的文件

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