如何高效存储PySpark查询结果？解决转Pandas/CSV过慢难题

1. 先做数据缩减，减少处理量 #

如果不需要全量数据，先在Spark SQL层面过滤或采样，从根源减少数据规模：

# 按时间范围过滤
df_filtered = spark.sql("""
SELECT id, start_time, end_time, value
FROM df_qck_dbase
WHERE start_time >= '2024-01-01' AND end_time <= '2024-06-01'
""")

# 或者随机采样（比如取10%数据，seed保证可复现）
df_sampled = df.sample(fraction=0.1, seed=42)

处理缩减后的数据集再转Pandas，速度会大幅提升。

2. 用列式存储格式（Parquet/ORC）替代CSV #

Parquet和ORC是专为大数据设计的列式存储格式，压缩率高、读取速度快，远优于CSV：

存储为Parquet #

# 直接保存为Parquet文件
df.write.mode("overwrite").parquet("/path/to/parquet_data")

# 按字段分区存储（比如按start_time分区，后续加载特定区间数据更高效）
df.write.mode("overwrite").partitionBy("start_time").parquet("/path/to/parquet_partitioned")

Pandas加载Parquet #

import pandas as pd
# 加载整个Parquet数据集
data = pd.read_parquet("/path/to/parquet_data")
# 加载分区数据时，直接读根目录即可，Pandas会自动识别分区

3. 分批次处理全量数据 #

如果必须处理全量数据，拆分批次避免一次性加载压力：

分批次转Pandas #

batch_size = 100000  # 根据内存情况调整批次大小
# 按分区拆分，逐个转成Pandas DataFrame
for idx, batch_df in enumerate(df.rdd.mapPartitions(lambda part: [pd.DataFrame(list(part))]).collect()):
    # 可以保存为单个小文件，或者逐步合并
    batch_df.to_csv(f"/path/to/batches/batch_{idx}.csv", index=False)

分批次存CSV #

# 每个CSV文件最多存10万条记录，自动生成多个文件
df.write.mode("overwrite")\
  .option("header", "true")\
  .option("maxRecordsPerFile", 100000)\
  .csv("/path/to/csv_batches")

后续用Pandas批量读取合并：

import glob
import pandas as pd

csv_files = glob.glob("/path/to/csv_batches/*.csv")
total_data = pd.concat([pd.read_csv(f) for f in csv_files], ignore_index=True)

4. 调优Spark配置提升性能 #

针对大任务调整Spark资源参数，优化并行度：

# 增大executor内存和核心数（根据集群资源调整）
spark.conf.set("spark.executor.memory", "8g")
spark.conf.set("spark.executor.cores", 4)

# 合并小分区，减少任务调度开销
df_repartitioned = df.repartition(10)  # 分区数根据数据量和集群节点数调整

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