Spark技术实现：如何获取时间区间内每日参数有效值？

实现步骤 #

要补全指定日期前每日的参数有效值（缺失日期沿用前序生效值），需完成以下核心操作：

1. 生成目标日期范围的连续日期序列：从所有参数的最早生效日期到指定日期，生成每一天的日期记录
2. 构建参数与日期的全量组合：获取所有唯一参数，与日期序列做笛卡尔关联，确保每个参数在目标区间内的每一天都有一条待填充记录
3. 匹配并填充每日有效值：将全量组合与原表关联，筛选每个参数在当日及之前的所有生效记录，通过窗口函数取最新的生效值作为当日有效值

完整代码实现 #

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

// 配置常量（根据实际场景调整）
val targetDate = "2022-01-06"
val validFromCol = "validfrom"
val compositeKey = Seq("param") // 对应你的CompositeKey，可扩展为多字段

// 1. 生成连续日期序列
val dateSeqDF = spark.table("table")
  .select(
    min(col(validFromCol)).cast("date").alias("start_date"),
    lit(targetDate).cast("date").alias("end_date")
  )
  .withColumn("dates", sequence($"start_date", $"end_date", expr("interval 1 day")))
  .select(explode($"dates").alias("calc_date"))

// 2. 获取所有唯一参数
val uniqueParamsDF = spark.table("table")
  .select(compositeKey.map(col): _*)
  .distinct()

// 3. 生成参数-日期全量组合
val paramDatePairsDF = uniqueParamsDF.crossJoin(dateSeqDF)

// 4. 关联原表数据，筛选每个参数在calc_date前的生效记录
val joinedDF = paramDatePairsDF.join(
  spark.table("table").select(
    compositeKey.map(col): _*,
    col(validFromCol).cast("date"),
    col("value")
  ),
  // 关联条件：参数匹配 + 生效日期<=计算日期
  compositeKey.map(k => col(k) === col(s"table.$k")) :+ (col(validFromCol) <= col("calc_date")): _*,
  "left"
)

// 5. 窗口函数取每个参数-日期对应的最新有效值
val windowSpec = Window.partitionBy(compositeKey.map(col): _*)
  .orderBy(col(validFromCol).desc)
  .rowsBetween(Window.unboundedPreceding, Window.currentRow)

val finalResultDF = joinedDF
  .withColumn("effective_value", first("value", ignoreNulls = true).over(windowSpec))
  .select(
    compositeKey.map(col): _*,
    col("calc_date").alias("date"),
    col("effective_value").alias("value")
  )
  .distinct()

finalResultDF.show()

关键说明 #

• 日期序列生成：利用sequence和explode函数快速生成连续日期，避免手动循环创建日期
• 笛卡尔关联：确保无遗漏的参数-日期组合，解决缺失日期的补全问题
• 窗口函数逻辑：按参数分区、生效日期倒序排列，使用first(..., ignoreNulls=true)直接取最新的非空生效值，高效完成前序值的沿用

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