pyspark reduce键是一个元组值嵌套列表

Question

我的问题如下：我正在解析用户交互，每次检测到交互时我都会发出 ((user1,user2),((date1,0),(0,1)))。这里的零代表交互的方向。

我无法弄清楚为什么我不能使用以下 reduce 函数来减少此输出：

def myFunc2(x1,x2):
    return (min(x1[0][0],x2[0][0]),max(x1[0][0],x2[0][0]),min(x1[0][1],x2[0][1]),max(x1[0][1],x2[0][1]),x1[1][0]+x2[1][0],x1[1][1]+x2[1][1])

我的映射器 (flatmap(myFunc)) 的输出是正确的：

((7401899, 5678002), ((1403185440.0, 0), (1, 0)))
((82628194, 22251869), ((0, 1403185452.0), (0, 1)))
((2162276, 98056200), ((1403185451.0, 0), (1, 0)))
((0509420, 4827510), ((1403185449.0, 0), (1, 0)))
((7974923, 9235930), ((1403185450.0, 0), (1, 0)))
((250259, 6876774), ((0, 1403185450.0), (0, 1)))
((642369, 6876774), ((0, 1403185450.0), (0, 1)))
((82628194, 22251869), ((0, 1403185452.0), (0, 1)))
((2162276, 98056200), ((1403185451.0, 0), (1, 0)))

但正在运行

lines.flatMap(myFunc) \
              .map(lambda x: (x[0], x[1])) \
              .reduceByKey(myFunc2)

给我错误

返回 (min(x1[0][0],x2[0][0]),max(x1[0][0],x2[0][0]),min(x1[0][ 1],x2[0][1]),max(x1[0][1],x2[0][1]),x1[1][0]+x2[1][0],x1[1 ][1]+x2[1][1])

TypeError: 'int' 对象没有属性 'getitem'

我想我在我的键中弄乱了一些东西，但我不知道为什么（我试图将键重铸为元组，如 here 所说但同样的错误）

有什么想法吗？非常感谢

Answer 1

好的，我认为这里的问题是您在没有您想象的那么深的项目中索引得太深。

我们来看看myFunc2

def myFunc2(x1,x2):
    return (min(x1[0][0],x2[0][0]),max(x1[0][0],x2[0][0]),min(x1[0][1],x2[0][1]),max(x1[0][1],x2[0][1]),x1[1][0]+x2[1][0],x1[1][1]+x2[1][1])

鉴于您的上述问题，输入数据将如下所示：

((467401899, 485678002), ((1403185440.0, 0), (1, 0)))

让我们继续并将该数据行分配给一个变量。

x = ((467401899, 485678002), ((1403185440.0, 0), (1, 0)))

当我们运行x[0] 时会发生什么？我们得到(467401899, 485678002)。当我们运行x[1]？我们得到((1403185440.0, 0), (1, 0))。这就是你的map 声明所做的，我相信。

好的。这很清楚。

在您的函数myFunc2 中，您有两个参数x1 和x2。这些对应于上面的变量：x1 = x[0] = (467401899, 485678002) 和 x2 = x[1] = ((1403185440.0, 0), (1, 0))

现在让我们只检查函数中return 语句的第一部分。

min(x1[0][0], x2[0][0])

所以，x1 = (467401899, 485678002)。凉爽的。现在，x1[0] 是什么？好吧，那是467401899。明显地。可是等等！ x1[0][0] 是什么？您尝试在x1[0] 处获取项目的第零个索引，但x1[0] 处的项目不是list 或tuple，它只是int。而<type 'int'> 的对象没有名为getitem 的方法。

总而言之：您对嵌套不那么深的对象挖掘得太深了。仔细考虑您传递给 myFunc2 的内容，以及您的对象有多深。

我认为myFunc2 的返回语句的第一部分应该如下所示：

return min(x1[0], x2[0][0])。您可以在 x2 上进行更深入的索引，因为 x2 具有更深的嵌套元组！

---

当我运行以下命令时，它工作得很好：

a = sc.parallelize([((7401899, 5678002), ((1403185440.0, 0), (1, 0))),
((82628194, 22251869), ((0, 1403185452.0), (0, 1))),
((2162276, 98056200), ((1403185451.0, 0), (1, 0))),
((1509420, 4827510), ((1403185449.0, 0), (1, 0))),
((7974923, 9235930), ((1403185450.0, 0), (1, 0))),
((250259, 6876774), ((0, 1403185450.0), (0, 1))),
((642369, 6876774), ((0, 1403185450.0), (0, 1))),
((82628194, 22251869), ((0, 1403185452.0), (0, 1))),
((2162276, 98056200), ((1403185451.0, 0), (1, 0)))])

b = a.map(lambda x: (x[0], x[1])).reduceByKey(myFunc2)

b.collect()

[((1509420, 4827510), ((1403185449.0, 0), (1, 0))),
 ((2162276, 98056200), (1403185451.0, 1403185451.0, 0, 0, 2, 0)),
 ((7974923, 9235930), ((1403185450.0, 0), (1, 0))), 
 ((7401899, 5678002), ((1403185440.0, 0), (1, 0))), 
 ((642369, 6876774), ((0, 1403185450.0), (0, 1))), 
 ((82628194, 22251869), (0, 0, 1403185452.0, 1403185452.0, 0, 2)),
 ((250259, 6876774), ((0, 1403185450.0), (0, 1)))]