python之Pyspark 应用不同的基于 reduce 函数的键

假设我有一些看起来像这样的数据

data =[('yes_sum', np.array([2, 2, 2])), 
 ('yes_sum', np.array([3, 3, 3])), 
 ('no_sum', np.array([4, 4, 4])), 
 ('no_sum', np.array([6, 6, 6]))] 

我将其转换为 rdd。

rdd_data = sc.parallelize(data) 

我想用键 'yes_sum' 对数组求和，但将键为 'no_sum' 的数组合并在一起。所以它看起来像这样:

[('yes_sum', array([5, 5, 5])), ('no_sum', array([4, 4, 4, 6, 6, 6]))] 

我只知道如何使用按键对数组求和:

rdd_data.reduceByKey(lambda x,y: x + y).collect() 

我得到:

[('yes_sum', array([5, 5, 5])), ('no_sum', array([10, 10, 10]))] 

但这不是我要找的。我在想这样的事情:

rdd_data.reduceByKey( 
    lambda x,y: if x.key() == 'yes_sum' x+y else np.concatenate((x, y)) 
).collect()  

请您参考如下方法:

首先，你的语法:

lambda x,y: if x.key() == 'yes_sum' x+y else np.concatenate((x, y)) 

不正确。相反，你可以这样写:

lambda x,y: x+y if x.key() == 'yes_sum' else np.concatenate((x, y)) 

但这会导致:

AttributeError: 'numpy.ndarray' object has no attribute 'key'

当您执行 reduceByKey 时，key 部分对于 reduce 函数本身是未知的。 Spark 已经完成了将类似键的数据分组在一起并将其传递给适当的缩减器的工作。

为了完成您想要做的事情，您需要在调用 reduceByKey 之前过滤 rdd。然后，您可以根据过滤应用不同的 reduce 函数，并合并您的结果。

例如:

yes_rdd = rdd_data.filter(lambda x: x[0] == 'yes_sum')\ 
    .reduceByKey(lambda x,y: x + y) 
 
no_rdd = rdd_data.filter(lambda x: x[0] != 'yes_sum')\ 
    .reduceByKey(lambda x,y: np.concatenate((x, y))) 
 
print(yes_rdd.union(no_rdd).collect()) 
#[('yes_sum', array([5, 5, 5])), ('no_sum', array([4, 4, 4, 6, 6, 6]))]