使用广播应用地图转换时，pyspark Udf 未按预期工作？

Question

我有如下两个列表

l=[['A', 'B', 'C'], ['A', 'C'], ['A', 'B', 'C'], ['A', 'B'],['B','C'],['B']]
x=[('A', 'B'), ('A', 'C')]

我想从列表l 列表中删除不包含列表x 中任何tuples 中的所有元素的所有元素。换句话说，x 中应该至少有一个tuple，其所有元组项都存在于l 的元素中。

根据我的last question，我在python中得到了以下解决方案：

print([l_ for l_ in l if any(all(e in l_ for e in x_) for x_ in x)])

产生所需的输出：

[['A', 'B', 'C'], ['A', 'C'], ['A', 'B', 'C'], ['A', 'B']]

现在我正在尝试使用 pyspark rdd 复制相同的操作，但我没有得到预期的结果。

这是我尝试过的：

rddsort=sc.parallelize(l)
broadcastVar = sc.broadcast(x)

def flist(unique_product_List,x):
    filter_list = [
        l_ for l_ in unique_product_List 
        if any(all(e in l_ for e in x_) for x_ in x)
    ]

    return filter_list

rddsort=rddsort.map(lambda flist(x[0],broadcastVar.value)) 
print(rddsort.collect())

结果我得到一个空列表列表：

[[], [], [], [], [], []]

但是我的预期结果应该和上面一样。

Answer 1

您需要对 rdd 进行过滤（不是地图）。过滤器将检查每一行的条件并删除不匹配的条件。这里的条件是行值 (list _l = l[0]) 应该包含 x 中的一个列表中的所有元素。

l=[['A', 'B', 'C'], ['A', 'C'], ['A', 'B', 'C'], ['A', 'B'],['B','C'],['B']]
x=[('A', 'B'), ('A', 'C')]
rddsort=sc.parallelize(l)

rddsort=rddsort.filter(lambda l_: any(all(e in l_ for e in x_) for x_ in x)) 
print(rddsort.collect())

输出

[['A', 'B', 'C'], ['A', 'C'], ['A', 'B', 'C'], ['A', 'B']]

更新： 在函数中使用广播变量：

l=[['A', 'B', 'C'], ['A', 'C'], ['A', 'B', 'C'], ['A', 'B'],['B','C'],['B']]
x=[('A', 'B'), ('A', 'C')]
rddsort=sc.parallelize(l)
broadcastVar = sc.broadcast(x)

def flist(row):
    filter_flag = any(all(e in l_ for e in x_) for x_ in broadcastVar.value)
    return filter_flag

rddsort=rddsort.filter(flist) 
print(rddsort.collect())