spark saveAsTextFile 最后一个分区（几乎？）永远不会完成

Question

我有一个非常简单的类似字数统计的程序，它可以生成（长，双）这样的字数：

val lines = sc.textFile(directory)

lines.repartition(600).mapPartitions{lineIterator => 
    // Generate iterator of (Long,Double) counts
}
.reduceByKey(new HashPartitioner(30), (v1, v2) => v1 + v2).saveAsTextFile(outDir, classOf[GzipCodec])

我的问题：30 个分区中的最后一个分区永远不会被写入。

这里有一些细节：

• 我的输入是 5 GB gz 压缩的，我预计大约 1B 的唯一长键。
• 我在 32 核 1.5TB 机器上运行。输入和输出来自具有 2TB 可用空间的本地磁盘。 Spark 被分配使用所有的 ram，并且很高兴地这样做了。此应用程序占用大约 0.5 TB。

我可以观察到以下几点：

• 对于 29 个分区，reduce 和 repartition（由于 HashPartitioner）大约需要 2 小时。最后一个没有完成，即使过了一天也没有。两到四个线程保持 100%。
• 日志中未出现错误或警告
• Spark 在 /tmp 中占用大约 100GB，这与 UI 报告的随机写入一致。
• 在 UI 中，我可以看到剩余任务的“随机读取记录”数量增长非常非常缓慢。一天后，距离所有已完成任务的显示还有一个数量级。

最后的日志是这样的：

15/08/03 23:26:43 INFO SparkHadoopWriter: attempt_201508031748_0002_m_000020_748: Committed
15/08/03 23:26:43 INFO Executor: Finished task 20.0 in stage 2.0 (TID 748). 865 bytes result sent to driver
15/08/03 23:27:50 INFO FileOutputCommitter: Saved output of task 'attempt_201508031748_0002_m_000009_737' to file:/output-dir/_temporary/0/task_201508031748_0002_m_000009
15/08/03 23:27:50 INFO SparkHadoopWriter: attempt_201508031748_0002_m_000009_737: Committed
15/08/03 23:27:50 INFO Executor: Finished task 9.0 in stage 2.0 (TID 737). 865 bytes result sent to driver
15/08/04 02:44:54 INFO BlockManager: Removing broadcast 3
15/08/04 02:44:54 INFO BlockManager: Removing block broadcast_3_piece0
15/08/04 02:44:54 INFO MemoryStore: Block broadcast_3_piece0 of size 2009 dropped from memory (free 611091153849)
15/08/04 02:44:54 INFO BlockManagerMaster: Updated info of block broadcast_3_piece0
15/08/04 02:44:54 INFO BlockManager: Removing block broadcast_3
15/08/04 02:44:54 INFO MemoryStore: Block broadcast_3 of size 3336 dropped from memory (free 611091157185)
15/08/04 02:44:54 INFO BlockManager: Removing broadcast 4
15/08/04 02:44:54 INFO BlockManager: Removing block broadcast_4_piece0
15/08/04 02:44:54 INFO MemoryStore: Block broadcast_4_piece0 of size 2295 dropped from memory (free 611091159480)
15/08/04 02:44:54 INFO BlockManagerMaster: Updated info of block broadcast_4_piece0
15/08/04 02:44:54 INFO BlockManager: Removing block broadcast_4
15/08/04 02:44:54 INFO MemoryStore: Block broadcast_4 of size 4016 dropped from memory (free 611091163496)

想象一下前五行在两分钟的时间范围内对 28 个其他分区重复。

我已经尝试了几件事：

• Spark 1.3.0 和 1.4.0
• nio 代替 netty
• flatMap 代替 mapPartitions
• 只有 30 个而不是 600 个输入分区

不过，我从来没有从火花中得到最后 1/30 的数据。有没有人观察到类似的东西？这两个帖子here 和here 似乎描述了类似的问题，但没有解决方案。

更新

永远不会完成的任务永远是reduceKey+writeToTextFile的第一个任务。我还删除了 HashPartitioner，甚至尝试在具有 400 个内核和 6000 个分区的更大集群上。只有 5999 成功完成，最后一个永远运行。

UI 显示所有任务，例如 随机读取大小/记录：20.0 MB / 1954832 但首先它显示（目前） 随机读取大小/记录：150.1 MB / 711836

数字仍在增长......

Answer 1

可能是您的密钥非常倾斜。根据它们的分布方式（或者如果您有一个空键或默认键），大量数据可能会发送到单个执行程序，并且与在本地机器上运行没有什么不同（加上分布式平台的开销）。它甚至可能导致该机器交换到磁盘，变得非常慢。

尝试使用aggregateByKey 而不是reduceByKey，因为它会尝试将部分总和分配给执行程序，而不是将所有（可能很大的）键值对集洗牌到单个执行程序。并且可能避免将输出分区的数量固定为 30，以防万一。

---

编辑：“它只是没有完成”很难检测到问题。您可以做的一件事是引入超时：

  val result = Await.result(future {
    // Your normal computation
  }, timeout)

这样，无论什么任务耗时过长，您都可以检测到它并当场收集一些指标。