如何用Apache Spark高效读取500GB CSV并完成列计算转换与ETL？

一、读取文件的核心优化 #

Spark自动推断Schema会全量扫描文件，这是读取慢的主要原因之一，必须手动指定Schema：

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

# 手动定义与CSV字段匹配的Schema
custom_schema = StructType([
    StructField("id", IntegerType(), nullable=False),
    StructField("value", DoubleType(), nullable=True),
    StructField("category", StringType(), nullable=True)
])

# 读取CSV，关闭Schema自动推断
df = spark.read.csv(
    "path/to/500gb.csv",
    header=True,
    schema=custom_schema,
    delimiter=",",
    quote='"',
    inferSchema=False
)

• 只加载需要的列：如果不需要全表字段，读取时直接通过select筛选，减少内存占用：

df = spark.read.csv(..., schema=custom_schema).select("id", "value", "category")

• 调整分区大小：默认每个分区128MB，500GB对应约4000个分区，可通过spark.sql.files.maxPartitionBytes调整（单位字节），避免分区过多或过少。

二、列计算与转换的高效实现 #

优先用Spark内置函数，避免Python UDF #

Python UDF会触发JVM与Python进程的序列化/反序列化，性能极差。直接用Spark SQL内置函数完成转换：

from pyspark.sql.functions import col, when, expr

# 示例：对value列计算平方，同时根据category打标签
df_transformed = df.withColumn(
    "value_squared", col("value") * col("value")
).withColumn(
    "tag", when(col("category") == "A", "high").otherwise("normal")
)

# 复杂计算可用expr写SQL表达式，更灵活
df_transformed = df.withColumn("value_squared", expr("value * value"))

必须用UDF时，用Pandas Vectorized UDF #

如果内置函数满足不了需求，用Pandas UDF（矢量化处理），性能比普通Python UDF高10-100倍：

from pyspark.sql.functions import pandas_udf
import pandas as pd
import numpy as np

# 定义Pandas UDF：计算value列的对数（假设value>0）
@pandas_udf("double")
def log_transform(s: pd.Series) -> pd.Series:
    return np.log(s)

df_transformed = df.withColumn("value_log", log_transform(col("value")))

三、聚合操作的性能优化 #

基础聚合用内置函数 #

直接用groupBy配合Spark内置聚合函数，避免自定义聚合逻辑：

from pyspark.sql.functions import sum, avg, count

# 按category聚合，计算value的总和、平均值和记录数
agg_result = df.groupBy("category").agg(
    sum("value").alias("total_value"),
    avg("value").alias("avg_value"),
    count("id").alias("record_count")
)

大分组场景优化Shuffle #

如果分组基数极大（比如百万级分组），调整Shuffle分区数，避免小分区过多：

# 设置Shuffle分区数与读取分区数匹配（比如4000）
spark.conf.set("spark.sql.shuffle.partitions", "4000")

# 开启自适应执行，让Spark自动调整Shuffle分区
spark.conf.set("spark.sql.adaptive.enabled", "true")

中间结果存列存格式 #

如果ETL需要多步操作，把中间结果存成Parquet/ORC格式（列存+压缩），后续读取速度会比CSV快数倍：

# 保存中间结果为Parquet
df_transformed.write.parquet("path/to/transformed_data", mode="overwrite")

# 后续读取直接用Parquet
df_loaded = spark.read.parquet("path/to/transformed_data")

四、通用性能调优 #

• 资源配置：如果是集群模式，给Executor足够的内存和CPU，比如：

spark-submit --executor-memory 16G --executor-cores 8 --num-executors 4 your_script.py

• 关闭不必要的检查：比如spark.sql.legacy.timeParserPolicy设置为LEGACY避免时间解析报错，根据实际需求调整。

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