在另一个 RDD 的基础上修剪一个 RDD

Question

spark 中是否有办法根据另一个 RDD 过滤一个 RDD 中的元素，即使它们不共享相同的键？

我有两个 RDD - abc 和 xyz

abc.collect() 看起来像这样

[[a,b,c],[a,c,g,e],[a,e,b,x],[b,c]]

xyz.collect() 看起来像这样

[a,b,c]

现在我想从 RDD abc 中过滤掉 xyz 中不存在的所有元素。

经过上述操作，RDD Abc 应该是这样的：

[[a,b,c],[a,c],[a,b],[b,c]]

我写了一个看起来像这样的代码：

def prune(eachlist):
    for i in eachlist:
        if i in xyz:
            return i

abc = abc.map(prune)

但是，这给了我这个错误：

Exception: It appears that you are attempting to broadcast an RDD or reference an RDD from an action or transformation. RDD transformations and actions can only be invoked by the driver, not inside of other transformations; for example, rdd1.map(lambda x: rdd2.values.count() * x) is invalid because the values transformation and count action cannot be performed inside of the rdd1.map transformation

我已经尝试过过滤器，查找而不是映射无济于事。我不断收到同样的错误。

我知道我可以在 xyz 上执行收集操作并解决此错误，但我在大型数据集上运行此操作并且执行 .collect() 会因为超出太多内存而杀死我的 AWS 服务器。因此，我需要在不使用 .collect() 或任何类似的昂贵操作的情况下执行此操作。

Answer 1

你可以：

# Add index
abc.zipWithIndex() \
    # Flatten values
    .flatMap(lambda x: [(k, x[1]) for k in x[0]]) \
    # Join with xyz (works as filter)
    .join(xyz.map(lambda x: (x, None))) \
    # Group back by index
    .map(lambda x: (x[1][0], x[0])) \
    .groupByKey() \
    .map(lambda x: list(x[1]))

或者您可以在xyz 上创建布隆过滤器并使用它来映射abc。