Spark技术需求：匹配DataFrame单行内的状态

看你提供的DataFrame结构，应该是要处理同一分组（Col1+Col2）内的状态序列匹配（比如追踪状态变化、识别特定状态流转组合）。下面基于Spark的DataFrame API和SQL窗口函数，给你几种实用的解决方案：

1. 用窗口函数追踪状态变化 #

如果你的需求是标记每一行状态与上一行的差异（比如判断当前状态是否切换），lag窗口函数是最直接的方式，它能帮你获取同一分组内上一行的状态：

Python 实现 #

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

# 初始化Spark会话
spark = SparkSession.builder.appName("StateMatchProcessing").getOrCreate()

# 定义窗口规则：按Col1、Col2分组，按时间Col3排序
window_spec = Window.partitionBy("Col1", "Col2").orderBy("Col3")

# 添加前一行状态列，生成状态变化标记
result_df = df.withColumn("prev_state", lag("Col4").over(window_spec)) \
              .withColumn("state_change", 
                          when(col("Col4") != col("prev_state"), "状态已切换")
                          .otherwise("状态未变化"))

# 查看结果
result_df.show(truncate=False)

Scala 实现 #

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.{lag, col, when}

val spark = SparkSession.builder.appName("StateMatchProcessing").getOrCreate()

val windowSpec = Window.partitionBy("Col1", "Col2").orderBy("Col3")

val resultDF = df.withColumn("prev_state", lag("Col4").over(windowSpec))
                 .withColumn("state_change", 
                             when(col("Col4") =!= col("prev_state"), "状态已切换")
                             .otherwise("状态未变化"))

resultDF.show(truncate = false)

2. 匹配特定状态流转组合 #

如果要找出符合特定顺序的状态对（比如Active之后紧跟Clear），可以用lead窗口函数获取下一行状态，再过滤出目标组合：

from pyspark.sql.functions import lead

# 获取下一行状态
target_df = df.withColumn("next_state", lead("Col4").over(window_spec)) \
              .filter((col("Col4") == "Active") & (col("next_state") == "Clear"))

# 输出所有Active→Clear的流转记录
target_df.show(truncate=False)

3. 处理复杂状态序列（比如连续Active） #

如果你的需求更复杂（比如识别连续3次Active的序列、自定义状态流转路径），可以用RDD的groupBy结合自定义函数来处理分组后的行迭代器：

from pyspark.sql.types import StructType, StructField, StringType, IntegerType, TimestampType

# 定义输出结果的Schema
output_schema = StructType([
    StructField("Col1", StringType()),
    StructField("Col2", IntegerType()),
    StructField("sequence_start", TimestampType()),
    StructField("sequence_end", TimestampType()),
    StructField("sequence_desc", StringType())
])

def process_state_group(group):
    # 拆分分组键和行迭代器
    (col1_val, col2_val), rows = group
    # 按时间排序行数据
    sorted_rows = sorted(rows, key=lambda r: r.Col3)
    
    sequence_list = []
    current_active_start = None
    
    for row in sorted_rows:
        if row.Col4 == "Active":
            # 记录连续Active的起始时间
            if not current_active_start:
                current_active_start = row.Col3
        else:
            # 遇到非Active状态，结束当前Active序列
            if current_active_start:
                sequence_list.append((
                    col1_val, col2_val,
                    current_active_start, row.Col3,
                    f"Active→{row.Col4}"
                ))
                current_active_start = None
    # 处理分组末尾未结束的Active序列
    if current_active_start:
        sequence_list.append((
            col1_val, col2_val,
            current_active_start, sorted_rows[-1].Col3,
            "Active(未结束)"
        ))
    return sequence_list

# 转换为RDD处理后再转回DataFrame
result_rdd = df.rdd.groupBy(lambda r: (r.Col1, r.Col2)).flatMap(process_state_group)
result_df = spark.createDataFrame(result_rdd, schema=output_schema)

result_df.show(truncate=False)

关键注意点 #

• 先确保Col3是TimestampType，如果是字符串类型需要先转换：df = df.withColumn("Col3", col("Col3").cast("timestamp"))
• 窗口函数适合大多数常规场景，性能更优；RDD自定义处理适合复杂业务逻辑，但要注意序列化问题

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