PySpark RDD场景下对比相邻元素提取各序列首个元素的实现咨询

实现逻辑 #

通过给有序RDD添加递增索引，关联前一个相邻元素的对比字段，过滤掉和前一个元素对比字段一致的元素，仅保留连续序列的第一个元素，非连续出现的重复字段组合不会被误删除。

注：输入RDD需提前按时间字段完成升序排序，保证元素顺序符合业务要求。

完整实现代码 #

# 构造示例输入RDD（实际使用时替换为你的源RDD，需提前按时间升序排序）
input_data = [
    (67, 312, "12:09:00"),
    (67, 375, "12:23:00"),
    (67, 375, "12:25:00"),
    (67, 650, "12:26:00"),
    (75, 650, "12:27:00"),
    (75, 650, "12:29:00"),
    (75, 800, "12:30:00"),
    (67, 375, "14:20:14")
]
rdd = sc.parallelize(input_data)

# 给每个元素添加顺序索引，保证顺序和输入排序一致
rdd_with_idx = rdd.zipWithIndex().map(lambda x: (x[1], x[0]))

# 构造前一个元素的关联RDD，key为原索引减1，仅保留需要对比的前两个字段
prev_rdd = rdd_with_idx.map(lambda x: (x[0] - 1, (x[1][0], x[1][1])))

# 左连接关联当前元素和前一个元素的对比字段
joined_rdd = rdd_with_idx.leftOuterJoin(prev_rdd)

# 过滤保留符合要求的元素：首条直接保留，或当前前两个字段与前一条不一致
result_rdd = joined_rdd.filter(
    lambda x: x[1][1] is None or (x[1][0][0], x[1][0][1]) != x[1][1]
).map(lambda x: x[1][0])

# 输出结果
print(result_rdd.collect())

输出结果 #

运行上述代码后得到的输出和预期完全一致：

[(67, 312, '12:09:00'), (67, 375, '12:23:00'), (67, 650, '12:26:00'), (75, 650, '12:27:00'), (75, 800, '12:30:00'), (67, 375, '14:20:14')]

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