PySpark中按顺序加载时序Parquet文件并保留顺序的问题

方法1：利用文件索引列实现低成本排序 #

因为每个Parquet文件内的数据已经是有序的，且文件本身按时间范围依次命名，我们可以给每个文件的数据添加一个文件顺序索引，再结合内部已有的时间戳排序，既能保留并行读取的效率，又能避免全量排序的内存压力：

from pyspark.sql import SparkSession
from pyspark.sql.functions import lit
from functools import reduce
from pyspark.sql import DataFrame

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

files = [
  '0to9999.parquet',
  '10000to19999.parquet',
  '20000to20000.parquet',
  ...
]

# 给每个文件的数据添加对应的顺序索引
dfs = []
for file_idx, file_path in enumerate(files):
    # 读取单个文件
    single_df = spark.read.parquet(file_path)
    # 添加文件索引列，标记该文件的顺序
    single_df = single_df.withColumn("file_order", lit(file_idx))
    dfs.append(single_df)

# 合并所有DataFrame
combined_df = reduce(DataFrame.union, dfs)

# 先按文件索引排序（保证文件顺序），再按时间戳排序（保证文件内顺序）
# 由于每个文件内已经有序，实际排序仅针对文件索引，成本极低
ordered_df = combined_df.orderBy("file_order", "你的时间戳列名").drop("file_order")

ordered_df.show()

方法2：强制串行读取并合并（适合文件数量少的场景） #

如果文件数量不多，且完全不需要并行读取，可以通过逐个读取并追加的方式保证顺序，但这种方式会牺牲读取效率：

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

# 初始化DataFrame为第一个文件的数据
ordered_df = spark.read.parquet(files[0])

# 逐个读取后续文件并追加
for file_path in files[1:]:
    temp_df = spark.read.parquet(file_path)
    # 使用unionByName保证列顺序一致（如果文件schema有变化时更安全）
    ordered_df = ordered_df.unionByName(temp_df)

ordered_df.show()

注意：这种方式下，Spark的执行计划会串行处理每个文件的读取与合并，因此整体速度会比并行读取慢，仅建议在文件数量较少时使用。

方法3：基于文件名提取时间范围排序 #

如果文件名本身包含明确的时间起始值（如示例中的0to9999），可以直接从文件名提取起始时间，以此作为排序依据，避免手动添加索引：

from pyspark.sql.functions import input_file_name, regexp_extract, cast

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

df = spark.read.parquet(*files)

# 从文件名中提取起始时间值（匹配数字开头的部分）
df = df.withColumn(
    "start_timestamp",
    regexp_extract(input_file_name(), r'^(\d+)to', 1).cast("long")
)

# 按起始时间排序，再按内部时间戳排序
ordered_df = df.orderBy("start_timestamp", "你的时间戳列名").drop("start_timestamp")

ordered_df.show()

这种方法无需提前遍历文件列表，直接利用文件名的规律实现排序，灵活性更高。

内容的提问来源于stack exchange，提问作者David Davó