首先，你当前用reduce(DataFrame.unionAll, torqueFilteredRawData)合并5600个DF慢到离谱，核心原因有两个：

1. 生成小DF的方式效率极低：你先把SD_GE_One的时间区间collect()到Driver端，再用map逐个筛选RawData，这不仅会把分布式数据拉到单点（极易引发OOM），还会生成5600个独立的小DF，每个DF都对应一次全表/分区扫描，重复计算量极大。
2. 逐个unionAll的合并逻辑不合理：Spark对大量小DF的连续union优化很差，每一次union都会触发shuffle或新stage，5600次操作会让查询计划变得臃肿不堪，调度开销远大于实际计算。

下面给你几个从根源到合并步骤的优化方案，按优先级排序：

方案1：从源头避免生成大量小DF（最优） #

完全抛弃collect()+map的逻辑，直接用Spark的分布式关联查询一次性筛选数据，这是效率最高的方式，所有操作都在集群上并行处理，没有单点瓶颈。

示例代码（Scala，PySpark语法类似） #

如果你的RawData中一条记录可能落在多个时间区间（需要去重），用exists子查询：

import org.apache.spark.sql.functions._

// 先确保时间列是Timestamp类型（字符串比较效率极低且易出错）
val rawDataWithTS = RawData.withColumn("SignalTimeStamp", to_timestamp(col("SignalTimeStamp")))
val timeRanges = SD_GE_One
  .withColumn("startTime", to_timestamp(col("startTime")))
  .withColumn("endTime", to_timestamp(col("endTime")))

val torqueFRData = rawDataWithTS.filter(
  exists(
    timeRanges,
    col("SignalTimeStamp").between(timeRanges("startTime"), timeRanges("endTime"))
  )
)

如果允许重复（同一条记录落在多个区间则保留多份），用join：

val torqueFRData = rawDataWithTS.join(
  timeRanges,
  rawDataWithTS("SignalTimeStamp").between(timeRanges("startTime"), timeRanges("endTime")),
  "inner"
).drop(timeRanges("startTime")).drop(timeRanges("endTime"))

为什么这更好？：Spark会自动优化这个查询计划，利用分区剪枝（如果RawData按SignalTimeStamp分区的话），避免重复扫描，全程没有单点数据拉取，计算完全并行。

方案2：如果必须保留小DF列表，优化合并逻辑 #

如果因为业务限制必须生成5600个DF，那不要逐个合并，而是分批合并，减少stage数量：

示例代码（Scala） #

val batchSize = 100 // 可根据集群资源调整，比如200-500区间都可以
// 把DF列表分成若干批次，每批次先合并成一个大DF
val batchedDFs = torqueFilteredRawData.grouped(batchSize).map { batch =>
  batch.reduce(_.unionByName(_)) // 用unionByName避免列顺序不一致的问题
}
// 最后合并所有批次的DF
val torqueFRData = batchedDFs.reduce(_.unionByName(_))

原理：一次性合并100个DF比合并5600个DF的查询计划简单得多，Spark能更好地优化批次内的合并，减少shuffle次数和调度开销。

方案3：额外优化项 #

• 检查分区策略：确保RawData按SignalTimeStamp分区，这样Spark在筛选时只会扫描符合时间范围的分区，避免全表扫描。
• 合并小文件：如果合并后生成大量小文件，可以执行torqueFRData.repartition(n)（n根据集群内存/磁盘带宽调整），减少文件数量，提升后续操作效率。
• 关闭不必要的日志：合并大量DF时，过多的日志会拖慢Driver速度，可以临时调低日志级别。

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