在 PySpark 中转换 RDD

【问题标题】：Transform RDD in PySpark在 PySpark 中转换 RDD
【发布时间】：2016-08-21 16:30:12
【问题描述】：

例如，我有下一个 ((i,j), k) 类型的 RDD：

((0,0), 0)
((0,1), 0)
((1,0), 0)
((1,1), 0)

我想将其转换为另一个，如果i==j 则具有1。我的第一次尝试出错了：

rddnew = rdd.flatMap(lambda ((i,j), k): [if i==j: ((i,j), 1)]))

有人可以帮我在 python 中改进这段代码吗？

【问题讨论】：

标签： apache-spark pyspark rdd

【解决方案1】：

这是一个解决方案：

data = [((0, 0), 0), ((0, 1), 0), ((1, 0), 0), ((1, 1), 0)]
rdd = sc.parallelize(data)
rdd2 = rdd.map(lambda ((i, j), _): ((i, j), 1) if (i == j) else ((i, j), 0))
rdd2.collect()
# [((0, 0), 1), ((0, 1), 0), ((1, 0), 0), ((1, 1), 1)]

您还可以通过使用映射器上的函数来定义更清洁的解决方案：

def transformation(entry):
    (i, j), v = entry
    return (i, j), v + 1 if i == j else 0

rdd3 = rdd.map(transformation)
rdd3.collect()
# [((0, 0), 1), ((0, 1), 0), ((1, 0), 0), ((1, 1), 1)]

【讨论】：