在Spark中按键减少到元组列表

Question

我试图转置我的数据，以便它是每个键的元组列表，而不是数据列。

作为我的投入：

1   234   54   7   9
2   654   34   2   1
1   987   22   4   6

我希望将输出写入拼花文件：

1:2   ((234, 54, 7, 9), (987, 22, 4, 6))
2:6   ((654, 34 2 1))

作为输入，我有两组拼花地板数据。我阅读它们并将其作为数据帧加入。然后我将每一行映射到键值对，然后将每个键还原为元组的大列表。

val keyedRDD = joinedDF.map(row => (
  ""+row.getInt(0)+":"+(row.getInt(1)/PARTITION_SIZE),
  List(Record(
    row.getInt(1),
    row.getInt(2),
    row.getInt(3),
    row.getInt(4)
  ))
))

val reduced = keyedRDD.reduceByKey(_:::_)

这里PARTITION_SIZE只是我为每次运行设置的一个变量，用于将数据分割成该大小的块。所以，如果我输入100000，并且有200万记录，那么编号为0-99,999的记录将在一个桶中，100,000-199,999将在下一个桶中，依此类推。

Record只是一个简单的case类来保存这些数据，我尝试过使用简单的元组，并将值单独放入一个列表中，得到了相同的结果。

我的理解是，这应该减少到每个键一个列表的输出，正如我前面所述。然而，我无法完成这项工作。在Spark History Server中，即使Ganglia显示至少80%的CPU使用率和高内存使用率，它也总是显示它在映射阶段挂起（甚至不启动它）。控制台被以下消息卡住：

16/01/18 01:26:10 INFO MapOutputTrackerMaster: Size of output statuses for shuffle 2 is 2485 bytes
16/01/18 01:26:10 INFO MapOutputTrackerMasterEndpoint: Asked to send map output locations for shuffle 2 to ip-172-31-7-127.ec2.internal:34337
16/01/18 01:26:10 INFO MapOutputTrackerMasterEndpoint: Asked to send map output locations for shuffle 2 to ip-172-31-7-129.ec2.internal:45407
16/01/18 01:26:17 INFO MapOutputTrackerMasterEndpoint: Asked to send map output locations for shuffle 1 to ip-172-31-7-128.ec2.internal:59468
16/01/18 01:26:17 INFO MapOutputTrackerMaster: Size of output statuses for shuffle 1 is 75087 bytes
16/01/18 01:26:18 INFO MapOutputTrackerMasterEndpoint: Asked to send map output locations for shuffle 1 to ip-172-31-7-127.ec2.internal:34337
16/01/18 01:26:18 INFO MapOutputTrackerMasterEndpoint: Asked to send map output locations for shuffle 1 to ip-172-31-7-129.ec2.internal:45407

一个数据集大约是3GB，另一个大约是22GB，所以一点都不大。但我想可能是我的内存不足了（即使我在被卡住20个小时后才收到OOM或executor丢失的消息）。我用m3试过EMR集群。xlarge带有2个从节点，m3。xlarge有6个从节点，甚至r3。xlarge有6个从节点，但仍然会遇到同样的问题。我已经设置了我的EMR集群，以便为Spark提供最大可能的内存分配、给定Spark动态分配、干扰memoryFraction设置等。

我只是不明白为什么这个会挂在那里。我试着简化它，只是在地图上做一个（键，1）RDD，并添加减少，它在20分钟内完成。

Answer 1

在昂贵的操作中附加到列表是一个常见的错误。记住Scala对不可变对象的偏见。最好从谷歌的“Scala列表附加性能”开始。这将为你提供几个很棒的博客，详细描述问题和建议。一般来说，添加到一个列表是一个昂贵的操作，因为每个操作都会产生一个新的列表——非常需要计算和内存。你可以在列表前加前缀，或者最好的答案通常是listbuffer。再次查看博客，并注意性能特征

http://www.scala-lang.org/docu/files/collections-api/collections_40.html