Spark 连接速度呈指数级缓慢

我正在尝试连接两个 Spark RDD。我有一个链接到类别的事务日志。我已将交易 RDD 格式化为以类别 id 作为键。

transactions_cat.take(3)
[(u'707', [u'86246', u'205', u'7', u'707', u'1078778070', u'12564', u'2012-03-02 00:00:00', u'12', u'OZ', u'1', u'7.59']), 
(u'6319', [u'86246', u'205', u'63', u'6319', u'107654575', u'17876', u'2012-03-02 00:00:00', u'64', u'OZ', u'1', u'1.59']), 
(u'9753', [u'86246', u'205', u'97', u'9753', u'1022027929', u'0', u'2012-03-02 00:00:00', u'1', u'CT', u'1', u'5.99'])]

categories.take(3)
[(u'2202', 0), (u'3203', 0), (u'1726', 0)]

事务日志约为 20 GB（3.5 亿行）。 类别列表小于1KB。

当我跑步时

transactions_cat.join(categories).count()

Spark 开始变得非常慢。我有一个有 643 个任务的阶段。前 10 个任务大约需要 1 分钟。然后每个任务的速度越来越慢（第 60 个任务大约需要 15 分钟）。我不确定出了什么问题。

Please check theses screenshots to get a better idea.

我正在使用 python shell 运行 Spark 1.1.0，有 4 个工作线程，总内存为 50 GB。 仅统计 RDD 的交易相当快（30 分钟）

问题可能在于 Spark 没有注意到您有一个简单的连接问题。当两者之一RDD你加入的公司太小了，所以最好不要成为一个RDD。然后你可以推出你自己的实现散列连接 http://en.wikipedia.org/wiki/Hash_join，这实际上比听起来简单得多。基本上，您需要：

• 将您的类别列表从RDD using collect()——由此产生的沟通将很容易收回成本（或者，如果可能的话，不要让它成为一个RDD首先）
• 将其转换为一个哈希表，其中一个条目包含一个键的所有值（假设您的键不是唯一的）
• 对于您的大号中的每一对RDD，在哈希表中查找键并为列表中的每个值生成一对（如果未找到，则该特定对不会生成任何结果）

我有一个Scala 中的实现 https://github.com/spirom/LearningSpark/blob/master/src/main/scala/special/HashJoin.scala——随意询问有关翻译的问题，但这应该很容易。

另一个有趣的可能性是尝试使用星火SQL https://spark.apache.org/sql/。我很确定该项目的长期目标将包括自动为您执行此操作，但我不知道他们是否已经实现了这一目标。