如何在PySpark中多线程并行读取列表中的多个分区文件？

关于多线程处理的疑问 #

不建议用Python线程来并行处理这3个文件。Spark本身是分布式计算框架，所有任务会自动分配到集群的Executor节点并行执行，Driver端手动开Python线程反而可能引发资源竞争、任务提交混乱等问题，不仅不会提升效率，还可能导致稳定性下降。

核心问题：批量读取时保留分区列 #

你的批量读取思路没问题，但不需要手动通过dates解析分区列——Spark默认会自动识别路径中的created_date=xxx、created_hour=xxx这类分区格式，直接将这两个字段作为列添加到DataFrame中。你之前手动用lit添加是多余的，反而可能因为DataFrame不可变特性（原代码中withColumn结果未赋值给df）导致无效。

优化后的批量处理方案 #

方案1：一次性读取所有目标分区（推荐）

如果集群资源足够，直接构造所有需要处理的分区路径，让Spark批量读取并统一处理，效率远高于逐个迭代：

def process_data(dates):
    # 构造所有目标分区的完整路径
    target_paths = [
        f's3://some_path/created_date={d[:10]}/created_hour={d[10:12]}'
        for d in dates
    ]
    # 批量读取，Spark自动识别created_date和created_hour列
    df = spark.read.orc(*target_paths)
    # 执行自定义转换逻辑（注意：Spark DataFrame不可变，需赋值返回结果）
    df = df.after_some_other_transformations()
    # 分区写入（repartition根据实际需求调整，若源数据分区合理可省略）
    df.repartition("created_date", "created_hour")\
      .write.mode('overwrite')\
      .partitionBy("created_date", "created_hour")\
      .format("orc")\
      .save(save_path)

方案2：分批次读取（超大数据量场景）

如果一次性读取所有分区内存压力过大，可以按日期分组批量处理（比如一个日期下的所有小时作为一批）：

from itertools import groupby

def process_data(dates):
    # 按created_date分组，同一日期的所有小时归为一批
    grouped_dates = groupby(dates, key=lambda x: x[:10])
    for date_str, hour_date_list in grouped_dates:
        # 构造当前批次的所有路径
        batch_paths = [
            f's3://some_path/created_date={date_str}/created_hour={d[10:12]}'
            for d in hour_date_list
        ]
        df = spark.read.orc(*batch_paths)
        df = df.after_some_other_transformations()
        df.repartition("created_date", "created_hour")\
          .write.mode('overwrite')\
          .partitionBy("created_date", "created_hour")\
          .format("orc")\
          .save(save_path)

原代码的错误修正 #

1. 方法1中withColumn未赋值给df，导致添加分区列的操作无效（Spark DataFrame是不可变对象，所有转换都需返回新对象）；
2. 方法2中循环变量错误：range(0, len(d), int(batch))应改为range(0, len(dates), batch)，file_array = dates[day:day+int(batch)]应改为file_array = dates[d:d+batch]。

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