基于Apache Spark的变更数据捕获：保留ID对应VALUE变更行的最优方案

这是个非常典型的时序数据清洗需求，用Spark的窗口函数就能高效解决，我给你拆解下最优方案：

核心思路 #

我们需要针对每个ID，按照时间顺序（DATE+TIME）对比相邻行的VALUE，只保留VALUE发生变化的行（包括每个ID的第一行，因为它没有前序对比项）。这里用LAG窗口函数来获取前一行的VALUE，然后过滤出符合条件的记录即可。

具体实现（Python版本） #

from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql.functions import lag, col

# 初始化SparkSession
spark = SparkSession.builder.appName("RetainValueChanges").getOrCreate()

# 读取数据集（示例用硬编码数据，实际可替换为CSV/Parquet等数据源）
sample_data = [
    ("001", "2019-01-01", "0010", 150),
    ("001", "2019-01-01", "0020", 150),
    ("001", "2019-01-01", "0030", 160),
    ("001", "2019-01-01", "0040", 160),
    ("001", "2019-01-01", "0050", 150),
    ("002", "2019-01-01", "0010", 151),
    ("002", "2019-01-01", "0020", 151),
    ("002", "2019-01-01", "0030", 161),
    ("002", "2019-01-01", "0040", 162),
    ("002", "2019-01-01", "0051", 152)
]
df = spark.createDataFrame(sample_data, ["ID", "DATE", "TIME", "VALUE"])

# 定义窗口规则：按ID分区，按DATE和TIME排序（保证时序正确）
window_spec = Window.partitionBy("ID").orderBy("DATE", "TIME")

# 添加前一行的VALUE字段
df_with_prev_value = df.withColumn("prev_value", lag("VALUE", 1).over(window_spec))

# 过滤条件：保留分区第一行（prev_value为null）或VALUE发生变化的行
result_df = df_with_prev_value.filter(
    col("prev_value").isNull() | (col("VALUE") != col("prev_value"))
).drop("prev_value")

# 查看结果
result_df.show()

关键细节说明 #

• 窗口分区与排序：partitionBy("ID")确保我们只在同一个ID的范围内做对比，orderBy("DATE", "TIME")保证按时间顺序取前一行，避免时序混乱导致错误。
• LAG函数：lag("VALUE",1)会获取当前行的上一行VALUE，每个ID的第一行没有前序行，所以prev_value为null，这行必须保留。
• 过滤逻辑：通过判断prev_value是否为null（第一行），或者当前VALUE与prev_value不等（值发生变化），就能精准筛选出你需要的记录。

生产环境优化建议 #

如果你的TIME字段格式不是标准的HHMM（比如存在类似010这种不补零的情况），建议先将DATE和TIME拼接成完整时间戳再排序，避免字符串排序错误：

from pyspark.sql.functions import concat, substring, to_timestamp

df = df.withColumn(
    "full_timestamp",
    to_timestamp(
        concat(col("DATE"), " ", substring(col("TIME"), 1, 2), ":", substring(col("TIME"), 3, 2)),
        "yyyy-MM-dd HH:mm"
    )
)
window_spec = Window.partitionBy("ID").orderBy("full_timestamp")

这样排序会更准确，尤其面对非标准格式的时间字符串时。

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