Spark 1.6中如何基于前一条记录更新DataFrame列

在Spark 1.6这个旧版本里处理这个需求确实得动点脑筋，毕竟很多高级API还没上线。我给你准备了两种可行的方案，你可以根据自己的数据规模和场景来选择：

这个方法通过在每个数据分区内部维护状态，记录最近一次col1=x时的col2和col3值，遇到col1=y的行就直接替换，处理效率比较高，适合大数据量场景。不过要注意数据必须先按正确的顺序排序（比如时间戳、自增ID），否则分区内的行顺序混乱会导致结果错误。

示例代码（Scala） #

假设你的DataFrame结构是(col1: String, col2: String, col3: String)，先确保数据已排序：

// 先按排序字段（比如时间戳id）排好序
val sortedDF = df.orderBy("id")

// 定义case class方便处理行数据（可选，但代码更清晰）
case class RowData(col1: String, col2: String, col3: String)

// 获取原DataFrame的schema，用于后续转换回DataFrame
val schema = sortedDF.schema

// 用mapPartitions处理每个分区的迭代器
val processedRDD = sortedDF.rdd.mapPartitions(iter => {
  // 初始化状态变量，存储最近一次x行的col2和col3
  var lastCol2: Option[String] = None
  var lastCol3: Option[String] = None
  
  iter.map(row => {
    val col1 = row.getString(0)
    val col2 = row.getString(1)
    val col3 = row.getString(2)
    
    col1 match {
      case "x" =>
        // 遇到x行，更新状态变量，保留原行数据
        lastCol2 = Some(col2)
        lastCol3 = Some(col3)
        Row(col1, col2, col3)
      case "y" =>
        // 遇到y行，用存储的状态值替换，若没有则保留原值（可根据需求调整）
        val newCol2 = lastCol2.getOrElse(col2)
        val newCol3 = lastCol3.getOrElse(col3)
        Row(col1, newCol2, newCol3)
      case _ =>
        // 其他类型的行，直接保留原数据
        Row(col1, col2, col3)
    }
  })
})

// 将处理后的RDD转换回DataFrame
val resultDF = sqlContext.createDataFrame(processedRDD, schema)

注意事项 #

• 如果x和y跨分区（比如前一个分区最后一行是x，下一个分区第一行是y），这个方法无法跨分区传递状态，此时需要先将数据repartition到单个分区（但大数据量下不推荐，会导致性能瓶颈）。
• 状态变量是每个分区独立维护的，所以必须保证排序后的行在分区内是连续的。

这个方法通过窗口函数给每个x行及后续的行分配同一个组ID，然后通过分组关联的方式把x行的col2/col3值填充到同组的y行里。优点是可以跨分区处理，但需要依赖一个可靠的排序字段，且全局窗口在大数据量下可能性能稍差。

示例代码（Scala） #

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

// 假设有一个排序字段（比如id），确保行的顺序正确
val windowSpec = Window.orderBy("id")

// 第一步：标记x行，计算分组ID
val withGroupDF = df
  .withColumn("is_x", when(col("col1") === "x", 1).otherwise(0))
  // 累计求和生成group_id，每个x行之后的行都属于同一个组，直到下一个x行
  .withColumn("group_id", sum("is_x").over(windowSpec))

// 第二步：提取每个分组对应的x行的col2和col3值
val groupValuesDF = withGroupDF
  .filter(col("col1") === "x")
  .select("group_id", "col2", "col3")
  .withColumnRenamed("col2", "x_col2")
  .withColumnRenamed("col3", "x_col3")

// 第三步：关联分组值，替换y行的col2和col3
val resultDF = withGroupDF
  .join(groupValuesDF, Seq("group_id"), "left")
  .withColumn("col2", when(col("col1") === "y", col("x_col2")).otherwise(col("col2")))
  .withColumn("col3", when(col("col1") === "y", col("x_col3")).otherwise(col("col3")))
  // 清理临时列
  .drop("is_x", "group_id", "x_col2", "x_col3")

注意事项 #

• 必须有一个可靠的排序字段（比如时间戳、自增ID），否则分组ID的计算会出错，导致填充值错误。
• 如果第一行是y，此时group_id为0，没有对应的x行，col2和col3会保留原值，你可以根据需求修改为null或者其他默认值。

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