使用 reducebykey 时出错：int 对象不可下标答案

【问题标题】：Error using reducebykey: int object is unsubscriptable使用 reducebykey 时出错：int 对象不可下标
【发布时间】：2018-06-24 18:49:08
【问题描述】：

执行以下脚本时出现错误“int object is unsubscriptable”：

element.reduceByKey( lambda x , y : x[1]+y[1])

with element 是一个key-value RDD，value是一个元组。示例输入：

(A, (toto , 10))
(A, (titi , 30))
(5, (tata, 10))
(A, (toto, 10))

我了解reduceByKey 函数采用 (K,V) 元组并对所有值应用函数以获得归约的最终结果。就像ReduceByKey Apache中给出的例子一样。

有什么帮助吗？

【问题讨论】：

你想要什么输出？问题是 x[1]+y[1] 是一个 int，而不是一个元组（这是 reduceByKey 在下一次迭代中所期望的。
预期的输出是(A , 50) (5, 10)，但为什么reduceByKey 应该在下一次迭代中期待一个元组呢？它应该保持相同类型的值减少吗？

标签： python apache-spark pyspark

【解决方案1】：

另一种方法是使用 Dataframe

rdd = sc.parallelize([('A', ('toto', 10)),('A', ('titi', 30)),('5', ('tata', 10)),('A', ('toto', 10))])
rdd.map(lambda (a,(b,c)): (a,b,c)).toDF(['a','b','c']).groupBy('a').agg(sum("c")).rdd.map(lambda (a,c): (a,c)).collect()

>>>[(u'5', 10), (u'A', 50)]

【讨论】：

【解决方案2】：

这里有一个例子来说明发生了什么。

让我们考虑一下当你在一个列表上调用reduce 时会发生什么f：

reduce(f, [a,b,c]) = f(f(a,b),c)

如果我们以f = lambda u, v: u[1] + v[1]为例，那么上面的表达式分解为：

reduce(f, [a,b,c]) = f(f(a,b),c) = f(a[1]+b[1],c)

但是a[1] + b[1] 是一个整数，所以没有__getitem__ 方法，因此你的错误。

一般来说，比较好的做法（如下图）是先用map()把数据提取成你想要的格式，然后再应用reduceByKey()。

包含您数据的 MCVE

element = sc.parallelize(
    [
        ('A', ('toto' , 10)),
        ('A', ('titi' , 30)),
        ('5', ('tata', 10)),
        ('A', ('toto', 10))
    ]
)

您可以几乎使用更复杂的 reduce 函数获得所需的输出：

def add_tuple_values(a, b):
    try:
        u = a[1]
    except:
        u = a
    try:
        v = b[1]
    except:
        v = b
    return u + v

print(element.reduceByKey(add_tuple_values).collect())

除了这会导致：

[('A', 50), ('5', ('tata', 10))]

为什么？ 因为键 '5' 只有一个值，所以没有什么可以减少的。

出于这些原因，最好先致电map。要获得所需的输出，您可以这样做：

>>> print(element.map(lambda x: (x[0], x[1][1])).reduceByKey(lambda u, v: u+v).collect())
[('A', 50), ('5', 10)]

更新 1

这里还有一种方法：

您可以在您的reduce 函数中创建tuples，然后调用map 以提取您想要的值。（基本上颠倒了map和reduce的顺序。）

print(
    element.reduceByKey(lambda u, v: (0,u[1]+v[1]))
        .map(lambda x: (x[0], x[1][1]))
        .collect()
)
[('A', 50), ('5', 10)]

备注

如果每个键至少有 2 条记录，使用 add_tuple_values() 会得到正确的输出。

【讨论】：