Pyspark Streaming：Windows与Linux CSV流处理行为差异及丢行问题

实际运行现象对比 #

两套独立部署的standalone Spark集群运行完全一致的代码，行为稳定分化：

• Windows环境（对应版本spark-3.1.1-bin-hadoop3.2）：配置maxFilesPerTrigger=1参数后，严格按CSV文件名的日期顺序，从最早的2010-12-01.csv开始逐次处理文件，全量处理完成后输出数千行聚合结果，符合预期
• Linux环境（对应版本PySpark 2.4.0.16，基于datastax docker镜像）：无固定顺序随机读取目录下的CSV文件，启动数分钟内就直接读取到最新日期的2011-12-10.csv，触发水位线阈值后判定剩余未读文件的数据为过期数据，最终仅输出41行结果就终止任务，多次运行结果无变化

核心配置代码 #

# Windows 端文件流初始化配置
streaming = spark.readStream.format("csv").schema(staticSchema)\
      .option("maxFilesPerTrigger", 1)\
      .load("D:\\data\\*.csv")

# Linux 端文件流初始化配置
streaming = spark.readStream.format("csv").schema(staticSchema)\
  .option("maxFilesPerTrigger", 1)\
  .load("file:///opt/data/*.csv")

# 两端通用的流处理、输出配置
stream = streaming.selectExpr("car", "cost", "timestamp")\
        .withWatermark("timestamp", "30 seconds")\
        .groupBy(F.col("car"), F.window("timestamp", "1 hour").alias("tmst_window"))\
        .agg(F.sum("cost").alias("agg_cost"))

stream.writeStream.format("console")\
  .queryName("customer_purchases")\
  .option('truncate', False)\
  .outputMode("append").start()

行为差异的根本原因 #

• 不同Spark版本的文件扫描排序逻辑差异
  Spark 2.4.0 版本的文件流数据源，扫描目录获取待处理文件列表时，直接沿用文件系统返回的原始条目顺序，没有做全局的字典序、时间戳排序。Linux环境下的ext4、xfs等主流文件系统，返回目录条目的顺序是基于inode分配的先后顺序，和文件名的日期逻辑没有绑定，因此会出现随机选到最新日期文件的情况。
  Spark 3.x 版本对该逻辑做了修正：扫描获取待处理文件后，会先对文件列表做全局排序（默认按文件名字典序），再按照maxFilesPerTrigger的配置逐批次分配处理，因此在Windows环境下才会表现为严格按日期从早到晚的顺序处理文件。
• 水位线提前推进的连锁影响
  因为Linux环境下随机扫描到了事件时间最新的2011-12-10.csv，流任务的全局水位线会直接推进到该文件的最大事件时间减去30秒的阈值。此时还未被扫描到的旧日期文件，所有事件时间都远低于当前水位线，会被直接判定为过期迟到数据，触发丢弃逻辑。
• 静态目录的任务终止判断逻辑
  Spark 2.4.0 版本的文件流，在判断任务是否可终止时，基于首次扫描到的目录快照：当扫描完成后如果没有检测到目录新增文件，就会判定当前目录是“静态无新增文件的数据集”，直接终止流任务，不会回头补扫之前因为随机顺序漏处理的旧文件，最终出现只处理了最新文件就结束任务的情况。

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