Spark按Instance分区替换前后值一致的连续Error块及统计需求

我是Spark新手，遇到以下数据处理问题多日未解决，寻求Java或Scala实现方案。

数据集 #

需求说明 #

需求1：Error值替换逻辑 #

按Instance分区，对连续的Error块：

• 若该块前后的非Error值相同，将块内所有Error替换为该值
• 若前后值不同，则保留Error

期望输出：

需求2：统计指标 #

统计每个Instance下各power_source的：

• 总数量
• 最大连续出现次数
• 最小连续出现次数

示例结果：

Electricity -> 总数量:5, 最大连续次数:5, 最小连续次数:5
Battery -> 总数量:3, 最大连续次数:3, 最小连续次数:3

目前需求1尚未解决，需求2计划将每个Instance的power_source收集为列表后通过UDF统计。

解决方案 #

需求1：Error替换实现（Scala） #

核心思路：通过窗口函数标记连续的Error块，再获取每个块的前后非Error值，判断后替换。

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

object SparkErrorReplace {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("ErrorReplace")
      .master("local[*]")
      .getOrCreate()
    import spark.implicits._

    // 构造示例数据
    val df = Seq(
      ("I1", "2024-09-01-00:00:00", "Electricity"),
      ("I1", "2024-09-01-00:15:00", "Error"),
      ("I1", "2024-09-01-00:30:00", "Error"),
      ("I1", "2024-09-01-00:45:00", "Error"),
      ("I1", "2024-09-01-01:00:00", "Electricity"),
      ("I1", "2024-09-01-01:15:00", "Error"),
      ("I1", "2024-09-01-01:30:00", "Battery"),
      ("I1", "2024-09-01-01:45:00", "Error"),
      ("I1", "2024-09-01-02:00:00", "Battery")
    ).toDF("Instance", "time", "power_source")

    // 1. 按Instance分区，time排序
    val windowSpec = Window.partitionBy("Instance").orderBy("time")

    // 2. 标记非Error的行，用于后续分组连续Error块
    val dfWithNonErrorFlag = df.withColumn(
      "is_non_error",
      when(col("power_source") =!= "Error", 1).otherwise(0)
    )

    // 3. 生成连续Error块的分组ID：累加非Error标记，相同组的Error属于同一块
    val dfWithBlockId = dfWithNonErrorFlag.withColumn(
      "block_id",
      sum("is_non_error").over(windowSpec.rangeBetween(Window.unboundedPreceding, 0))
    )

    // 4. 获取每个block的前一个非Error值（向前填充非Error值）
    val dfWithPrevValue = dfWithBlockId.withColumn(
      "prev_value",
      last(when(col("power_source") =!= "Error", col("power_source")), ignoreNulls = true)
        .over(windowSpec.rangeBetween(Window.unboundedPreceding, -1))
    )

    // 5. 获取每个block的后一个非Error值（向后填充非Error值）
    val dfWithNextValue = dfWithPrevValue.withColumn(
      "next_value",
      first(when(col("power_source") =!= "Error", col("power_source")), ignoreNulls = true)
        .over(windowSpec.rangeBetween(1, Window.unboundedFollowing))
    )

    // 6. 替换Error值：前后值相同则替换，否则保留Error
    val resultDf = dfWithNextValue.withColumn(
      "power_source",
      when(
        col("power_source") === "Error" && col("prev_value") === col("next_value"),
        col("prev_value")
      ).otherwise(col("power_source"))
    ).drop("is_non_error", "block_id", "prev_value", "next_value")

    resultDf.show(false)
  }
}

Java版本思路

Java实现逻辑与Scala完全一致，核心是通过Window类定义窗口，调用functions静态方法使用when、sum、last、first等函数，最终完成条件替换，可参考Scala代码步骤逐步转换。

需求2：统计指标实现（Scala） #

无需收集列表，直接通过窗口函数标记连续相同的power_source分组，再聚合统计：

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

object SparkSourceStats {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("SourceStats")
      .master("local[*]")
      .getOrCreate()
    import spark.implicits._

    // 使用需求1处理后的结果作为输入
    val resultDf = Seq(
      ("I1", "2024-09-01-00:00:00", "Electricity"),
      ("I1", "2024-09-01-00:15:00", "Electricity"),
      ("I1", "2024-09-01-00:30:00", "Electricity"),
      ("I1", "2024-09-01-00:45:00", "Electricity"),
      ("I1", "2024-09-01-01:00:00", "Electricity"),
      ("I1", "2024-09-01-01:15:00", "Error"),
      ("I1", "2024-09-01-01:30:00", "Battery"),
      ("I1", "2024-09-01-01:45:00", "Battery"),
      ("I1", "2024-09-01-02:00:00", "Battery")
    ).toDF("Instance", "time", "power_source")

    // 1. 按Instance分区，time排序
    val windowSpec = Window.partitionBy("Instance").orderBy("time")

    // 2. 标记连续相同的power_source分组：当前值与前一个不同时，分组ID+1
    val dfWithGroup = resultDf.withColumn(
      "group_id",
      sum(when(col("power_source") =!= lag("power_source", 1).over(windowSpec), 1).otherwise(0))
        .over(windowSpec.rangeBetween(Window.unboundedPreceding, 0))
    )

    // 3. 统计每个分组的连续次数
    val groupStats = dfWithGroup.groupBy("Instance", "power_source", "group_id")
      .agg(count("*").alias("continuous_count"))

    // 4. 聚合得到总数量、最大/最小连续次数
    val finalStats = groupStats.groupBy("Instance", "power_source")
      .agg(
        sum("continuous_count").alias("总数量"),
        max("continuous_count").alias("最大连续次数"),
        min("continuous_count").alias("最小连续次数")
      )

    // 格式化输出结果
    finalStats.select(
      concat(
        col("power_source"),
        lit(" -> 总数量:"), col("总数量"),
        lit(", 最大连续次数:"), col("最大连续次数"),
        lit(", 最小连续次数:"), col("最小连续次数")
      ).alias("统计结果")
    ).show(false)
  }
}

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