时间戳

首页 > 解决方案 > 如何在pyspark中使用其他Rdd元素的所有可能组合创建新Rdd？

python-3.x - 如何在pyspark中使用其他Rdd元素的所有可能组合创建新Rdd？

问题描述

嗨，我创建了一个如下所示的 Rdd

rdd1=sc.parallelize(['P','T','K'])
rdd1.collect()
['P', 'T', 'K']

现在我想RDD2用所有可能的组合创建新的，如下面的新RDD.ie除了相同的元素组合，如（p，p），（k，k），（t，t）。

我在做时的预期输出

RDD2.collect()

[
    ('P'),('T'),('K'),
    ('P','T'),('P','K'),('T','K'),('T','P'),('K','P'),('K','T'),
    ('P','T','K'),('P','K','T'),('T','P','K'),('T','K','P'),('K','P','T'),('K','T','P')
]

标签： python-3.xapache-sparkpysparkrdd

解决方案

您似乎希望rdd在每行包含唯一值的地方生成元素的所有排列。

一种方法是首先创建一个辅助函数来生成所需的长度组合n：

from functools import reduce
from itertools import chain

def combinations_of_length_n(rdd, n):
    # for n > 0
    return reduce(
        lambda a, b: a.cartesian(b).map(lambda x: tuple(chain.from_iterable(x))),
        [rdd]*n
    ).filter(lambda x: len(set(x))==n)

本质上，该函数将自己进行n笛卡尔积，rdd并仅保留所有值都不同的行。

我们可以对此进行测试n = [2, 3]：

print(combinations_of_length_n(rdd1, n=2).collect())
#[('P', 'T'), ('P', 'K'), ('T', 'P'), ('K', 'P'), ('T', 'K'), ('K', 'T')]

print(combinations_of_length_n(rdd1, n=3).collect())
#[('P', 'T', 'K'),
# ('P', 'K', 'T'),
# ('T', 'P', 'K'),
# ('K', 'P', 'T'),
# ('T', 'K', 'P'),
# ('K', 'T', 'P')]

您想要的最终输出只是union这些中间结果与原始结果rdd（值映射到tuples）。

rdd1.map(lambda x: tuple((x,)))\
    .union(combinations_of_length_n(rdd1, 2))\
    .union(combinations_of_length_n(rdd1, 3)).collect()
#[('P',),
# ('T',),
# ('K',),
# ('P', 'T'),
# ('P', 'K'),
# ('T', 'P'),
# ('K', 'P'),
# ('T', 'K'),
# ('K', 'T'),
# ('P', 'T', 'K'),
# ('P', 'K', 'T'),
# ('T', 'P', 'K'),
# ('K', 'P', 'T'),
# ('T', 'K', 'P'),
# ('K', 'T', 'P')]

概括任何最大重复次数：

num_reps = 3
reduce(
    lambda a, b: a.union(b),
    [
        combinations_of_length_n(rdd1.map(lambda x: tuple((x,))), i+1) 
        for i in range(num_reps)
    ]
).collect()
#Same as above

注意：笛卡尔积是昂贵的操作，应尽可能避免。

推荐阅读