使用PySpark读取自定义扩展名的GZIP压缩CSV文件

Spark默认通过文件扩展名推断压缩编码，你的文件实际是GZIP压缩但后缀是.csv，所以会被当作未压缩文本解析，导致失败。以下两种Python方案可以解决这个问题：

方案1：强制指定压缩格式 #

直接在读取CSV时明确指定压缩编码为gzip，同时配置CSV的相关参数：

from pyspark.sql import SparkSession

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

# 读取文件（根据实际情况调整参数）
df = spark.read \
    .option("compression", "gzip") \
    .option("header", "true")  # 你的文件有表头就保留，没有则删除
    .option("inferSchema", "false")  # 推荐手动指定Schema，提升性能
    .csv("s3://your-bucket/path/to/target-files/*.csv")

# 示例：手动指定Schema（替换成你的字段结构）
from pyspark.sql.types import StructType, StructField, StringType, DoubleType

custom_schema = StructType([
    StructField("user_id", StringType(), nullable=True),
    StructField("order_amount", DoubleType(), nullable=True),
    StructField("order_date", StringType(), nullable=True)
])

# 使用自定义Schema读取
df_with_schema = spark.read \
    .option("compression", "gzip") \
    .option("header", "true") \
    .schema(custom_schema) \
    .csv("s3://your-bucket/path/to/target-files/*.csv")

# 验证结果
df_with_schema.show(5)

核心是option("compression", "gzip")，它会强制Spark忽略文件扩展名，用GZIP解码内容。

方案2：底层Hadoop输入格式控制 #

如果方案1不生效，可直接指定Hadoop的GZIP压缩解码器和文本输入格式：

from pyspark.sql import SparkSession

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

df = spark.read \
    .format("csv") \
    .option("header", "true") \
    .option("inputFormatClass", "org.apache.hadoop.mapreduce.lib.input.TextInputFormat") \
    .option("compressionCodec", "org.apache.hadoop.io.compress.GzipCodec") \
    .load("s3://your-bucket/path/to/target-files/*.csv")

这种方式直接绕过扩展名推断，强制Hadoop用GZIP解压后再交给CSV解析器处理。

关键注意事项 #

• 确保Spark集群已配置S3访问权限（比如绑定IAM角色、设置AWS_ACCESS_KEY_ID等），否则会出现权限报错。
• 尽量避免用inferSchema，大文件下会触发全量扫描，严重影响性能，手动指定Schema更高效。

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