Spark Scala DataFrame：提取连续10分钟status为1的小时批次首行

每小时从HBase表加载原始数据至Spark Scala DataFrame，需实现以下逻辑：

• 针对每个ID，若status连续10分钟以上保持为1，则提取对应小时批次中的首行数据
• 若在10分钟阈值内status发生变化，则无需提取该行

示例输入DataFrame #

预期输出 #

反例说明 #

若10分钟阈值内status发生变化，则无数据提取，示例输入如下：

实现思路 #

1. 将字符串类型的Timestamp转换为Spark时间戳类型，方便时间计算
2. 按ID和小时维度分组，匹配每小时的批次数据
3. 对每个分组内的数据按时间排序，划分连续status=1的区间
4. 计算每个连续区间的时长，判断是否满足10分钟阈值，若满足则提取该分组的首行

代码实现 #

import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.TimestampType
import org.apache.spark.sql.Window

// 假设原始加载的DataFrame名为rawDF，结构为：ID: Int, Timestamp: String, status: Int
val resultDF = rawDF
  // 1. 转换时间字符串为Spark时间戳类型
  .withColumn("ts", col("Timestamp").cast(TimestampType))
  // 2. 提取小时维度，用于按小时批次分组
  .withColumn("hour_batch", date_trunc("hour", col("ts")))
  // 3. 给每个小时批次内的行按时间排序，标记行号
  .withColumn("row_seq", row_number().over(Window.partitionBy("ID", "hour_batch").orderBy("ts")))
  // 4. 生成连续status的分组标识：当status变化时，分组编号+1
  .withColumn("status_group", 
    sum(when(col("status") !== lag("status", 1, col("status")).over(Window.partitionBy("ID", "hour_batch").orderBy("ts")), 1)
      .otherwise(0)).over(Window.partitionBy("ID", "hour_batch").orderBy("ts"))
  )
  // 5. 过滤出status=1的记录，计算每个连续分组的时间跨度
  .filter(col("status") === 1)
  .withColumn("interval_start", min("ts").over(Window.partitionBy("ID", "hour_batch", "status_group")))
  .withColumn("interval_end", max("ts").over(Window.partitionBy("ID", "hour_batch", "status_group")))
  .withColumn("duration_min", (unix_timestamp(col("interval_end")) - unix_timestamp(col("interval_start")))/60)
  // 6. 判断当前小时批次是否存在满足10分钟的连续status=1区间
  .withColumn("meet_threshold", max(col("duration_min")).over(Window.partitionBy("ID", "hour_batch")) >= 10)
  // 7. 提取满足条件的小时批次的首行数据
  .filter(col("meet_threshold") && col("row_seq") === 1)
  // 8. 保留原始需求的列
  .select("ID", "Timestamp", "status")

// 查看结果
resultDF.show()

代码说明 #

• 小时批次划分：通过date_trunc提取小时维度，确保只在每小时的批次内处理数据
• 连续状态分组：用lag函数对比前一行的status，生成连续状态的分组标识，区分不同的连续区间
• 时长计算：计算每个连续status=1区间的起止时间差，转换为分钟级时长判断是否达标
• 首行提取：标记每个小时批次内的行号，仅保留满足阈值的批次的第一行

结果验证 #

• 示例输入中，10点批次的status连续保持1，时长超过10分钟，因此提取首行；11点批次的连续时长仅5分钟，不满足阈值，无输出
• 反例输入中，两个小时批次的status都在10分钟内发生变化，连续时长不足，因此无输出

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