首页 文章

PySpark，Key的交叉点

提问于
浏览 次
1

例如，我在PySpark中有两个RDD：

((0,0), 1)
((0,1), 2)
((1,0), 3)
((1,1), 4)

第二是

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

我希望从第一个RDD到第二个RDD的交叉点 . 实际上，第二个RDD必须扮演第一个掩模的角色 . 输出应该是：

((0,1), 2)
((1,1), 4)

它表示来自第一个RDD的值，但仅表示来自第二个RDD的键 . 两个RDD的长度不同 .

我有一些解决方案（必须证明），但是这样的事情：

rdd3 = rdd1.cartesian(rdd2)
rdd4 = rdd3.filter(lambda((key1, val1), (key2, val2)): key1 == key2)
rdd5 = rdd4.map(lambda((key1, val1), (key2, val2)): (key1, val1))

我不知道，这个解决方案的效率如何 . 想听听经验丰富的Spark程序员的意见....

pyspark rdd

1 回答

  • 2

    也许我们不应该把这个过程想象为加入 . 你真的不想加入两个数据集，你想要从另一个数据集中减去一个数据集吗？

    我将从你的问题中说明我的假设

    • 您根本不关心第二个数据集中的值 .

    • 您只想将值保留在第一个数据集中，其中键值对出现在第二个数据集中 .

    Idea 1 ：Cogroup（我想可能是最快的方式） . 它基本上是计算两个数据集的交集 .

    rdd1 = sc.parallelize([((0,0), 1), ((0,1), 2), ((1,0), 3), ((1,1), 4)])
rdd2 = sc.parallelize([((0,1), 3), ((1,1), 0)])
intersection = rdd1.cogroup(rdd2).filter(lambda x: x[1][0] and x[1][1])
final_rdd = intersection.map(lambda x: (x[0], list(x[1][0]))).map(lambda (x,y): (x, y[0]))

    Idea 2 ：按键减去

    rdd1 = sc.parallelize([((0,0), 1), ((0,1), 2), ((1,0), 3), ((1,1), 4)])
rdd2 = sc.parallelize([((0,1), 3), ((1,1), 0)])

unwanted_rows = rdd1.subtractByKey(rdd2)
wanted_rows = rdd1.subtractByKey(unwanted_rows)

    如果这比你的方法更快，我不是100％肯定 . 它确实需要两个 subtractByKey 操作，这可能很慢 . 此外，此方法不保留顺序（例如， ((0, 1), 2) ，尽管在第一个数据集中是第一个，但在最终数据集中是第二个） . 但我无法想象这很重要 .

    至于哪个更快，我认为这取决于你的卡特尔加入需要多长时间 . 映射和过滤往往比 subtractByKey 所需的shuffle操作更快，但当然 cartesian 是一个耗时的过程 .

    无论如何，我想你可以尝试这种方法，看看它是否适合你！

    性能改进的附注，取决于您的RDD的大小 .

    如果 rdd1 足够小，可以保存在主存储器中，如果你广播它，然后对它进行流 rdd2 ，则可以大大加快减法过程 . 但是，我承认这种情况很少发生 .

    回复于

相关问题