基于Spark与Scala合并不同频率时序DataFrame的实现方案

针对你提出的两种数据量场景，分别给出高效实现方案：

场景1：Value1数据量极大，Value2仅数百条 #

这种场景下，直接把小体量的Value2数据集加载到Driver端内存，通过自定义UDF匹配即可，完全避免分布式shuffle操作，效率极高。

实现步骤&代码 #

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

// 构造示例数据（实际替换为你的业务DataFrame）
val df1 = spark.createDataFrame(Seq(
  (10, 100),
  (20, 200),
  (30, 300)
)).toDF("relative_time", "value1")

val df2 = spark.createDataFrame(Seq(
  (1, 10),
  (22, 220)
)).toDF("relative_time", "value2")

// 将df2按时间降序排序后收集到Driver端，方便快速查找最新前置值
val sortedValue2List = df2.orderBy(desc("relative_time"))
  .collect()
  .map(row => (row.getAs[Int]("relative_time"), row.getAs[Int]("value2")))

// 定义UDF：输入当前时间，返回Value2中不大于该时间的最新值
val getLatestValue2 = udf((currentTime: Int) => {
  sortedValue2List.find(_._1 <= currentTime).map(_._2)
})

// 给df1添加匹配到的value2列
val finalResult = df1.withColumn("value2", getLatestValue2(col("relative_time")))
finalResult.show()

说明 #

因为Value2只有数百条，收集到Driver端的内存开销可以忽略；UDF内的查找是线性遍历，但数据量极小，不会成为性能瓶颈。

场景2：两者数据量均极大 #

这种场景必须用分布式原生操作避免内存溢出，优先使用Spark 3.0+引入的ASOF Join（专门为时序异步关联设计），低版本Spark可以用区间Join+窗口函数的优化方案。

方案1：ASOF Join（Spark 3.0+推荐） #

ASOF Join的核心逻辑就是：为左表（规则采样的df1）的每一行，匹配右表（不规则的df2）中键值不大于左表对应键的最新行，完全贴合需求，且性能远优于自定义Join+窗口。

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

// 必须保证两张表都按关联时间字段排序（ASOF Join要求）
val sortedDf1 = df1.orderBy("relative_time")
val sortedDf2 = df2.orderBy("relative_time")

// 执行ASOF Join
val finalResult = sortedDf1.asOfJoin(
  sortedDf2,
  leftKeys = Seq("relative_time"),
  rightKeys = Seq("relative_time"),
  condition = lit(true) // 若有分组维度（比如多设备时序），可在此添加分组匹配条件
).select(
  sortedDf1("relative_time"),
  sortedDf1("value1"),
  sortedDf2("value2")
)

finalResult.show()

方案2：区间Join+窗口函数（兼容Spark 3.0以下） #

如果无法升级Spark版本，可通过两次Join缩小中间数据量，避免全量窗口计算的性能损耗：

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

// 先对df2去重，保留每个时间点的最新value2（若df2存在重复时间的情况）
val deduplicatedDf2 = df2.groupBy("relative_time")
  .agg(last("value2").alias("value2"))

// 第一步：区间Join，获取所有符合时间条件的df2记录
val joinedTemp = df1.join(
  deduplicatedDf2,
  deduplicatedDf2("relative_time") <= df1("relative_time"),
  "left"
)

// 第二步：按df1的时间分组，找到每个时间点对应的最大df2时间，再关联回value2
val finalResult = joinedTemp.groupBy(df1("relative_time"), df1("value1"))
  .agg(max(deduplicatedDf2("relative_time")).alias("latest_rt"))
  .join(deduplicatedDf2, col("latest_rt") === deduplicatedDf2("relative_time"), "left")
  .select("relative_time", "value1", "value2")

finalResult.show()

说明 #

当Value2采样频率远高于Value1时，ASOF Join的优势尤为明显——它利用数据的有序性直接定位匹配行，无需处理大量冗余的Value2数据，shuffle开销极低。

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