SQL游标转Spark DataFrame实现咨询：自连接行比较与条件更新逻辑转换

嘿，我来帮你把这个SQL游标逻辑转成Spark DataFrame的实现方式～首先得先拆解清楚原SQL的核心逻辑：它其实是在做同表自匹配——遍历每一行，找到和它同CustomerRef、同CompCode且日期范围重叠的其他行，然后把两行的ComplaintStatusRefCode转换成对应的数值（OldValue和NewValue），后续应该是要基于这两个值做更新操作对吧？

Spark里完全不需要用游标这种逐行处理的方式（毕竟Spark是分布式批量计算，游标效率极低），用DataFrame的自连接、条件过滤和列计算就能搞定，步骤如下：

1. 给原DataFrame加唯一标识，避免自我匹配 #

因为是自连接，得先给每一行加个唯一ID，防止某一行和自己匹配上：

from pyspark.sql.functions import monotonically_increasing_id, col, when, coalesce, row_number

# 假设你的原始数据DataFrame名为raw_df
df_with_id = raw_df.withColumn("row_id", monotonically_increasing_id())

2. 执行自连接，匹配符合条件的行 #

这一步对应你SQL里的IF条件，把原表和自己连接，只保留满足匹配规则的行：

# 用别名区分原行（a）和待比较的行（b）
joined_df = df_with_id.alias("a").join(
    df_with_id.alias("b"),
    # 匹配相同的CustomerRef和CompCode
    (col("a.CustomerRef") == col("b.CustomerRef")) &
    (col("a.CompCode") == col("b.CompCode")) &
    # 日期范围重叠逻辑，用coalesce处理NULL值（对应SQL里的ISNULL）
    (col("a.StartDate") <= coalesce(col("b.EndDate"), col("b.ProposedEndDate"))) &
    (coalesce(col("a.EndDate"), col("a.ProposedEndDate")) >= col("b.StartDate")) &
    # 排除行自身匹配的情况
    (col("a.row_id") != col("b.row_id")),
    how="inner"
)

3. 计算OldValue和NewValue #

对应你SQL里的CASE语句，用Spark的when函数实现状态到数值的映射：

# 给原行（a）计算OldValue，给匹配行（b）计算NewValue
mapped_df = joined_df.withColumn("OldValue", 
    when(col("a.ComplaintStatusRefCode") == "LODGED", 4)
    .when(col("a.ComplaintStatusRefCode") == "FINISHED", 3)
    .when(col("a.ComplaintStatusRefCode") == "REPORTED", 2)
    .otherwise(1)
).withColumn("NewValue",
    when(col("b.ComplaintStatusRefCode") == "LODGED", 4)
    .when(col("b.ComplaintStatusRefCode") == "FINISHED", 3)
    .when(col("b.ComplaintStatusRefCode") == "REPORTED", 2)
    .otherwise(1)
)

4. 基于匹配结果更新原数据（按需选择） #

如果你的最终需求是更新原DataFrame里的记录（比如保留每个客户/公司组里状态值最高的记录），可以用窗口函数来实现：

from pyspark.sql.window import Window

# 按CustomerRef和CompCode分组，按NewValue降序排序
window_spec = Window.partitionBy("a.CustomerRef", "a.CompCode").orderBy(col("NewValue").desc())

# 取每个分组里NewValue最大的那条记录作为最终结果
final_df = mapped_df.withColumn("row_rank", row_number().over(window_spec)) \
    .filter(col("row_rank") == 1) \
    .select("a.*", "NewValue")  # 这里选择你需要保留的列

最后补充一句：Spark的这种批量处理方式比游标高效太多了，尤其是数据量较大的时候，完全适配分布式计算的特性，不会出现逐行处理的性能瓶颈。

备注：内容来源于stack exchange，提问作者Pysparker