这是一篇来自Facebook的留言。Facebook工程团队的软件工程师tejaspatil和jingzheng展示了如何使用apachespark进行大规模建模。该博客最初发布在Facebook代码博客上。处理大规模数据是Facebook数据基础设施集团的核心工作。这些年来，我们看到我们的分析需求有了巨大的增长，为了满足这些需求，我们要么设计并构建一个新的系统，要么采用现有的开源解决方案并加以改进，使之能够在我们的规模内发挥作用。对于我们的一些批处理用例，我们决定使用apachespark，这是一个快速增长的开源数据处理平台，能够扩展大量数据并支持定制用户应用程序。几个月前，我们共享了一个这样的用例，它利用了Spark的声明性（SQL）支持。在这篇文章中，我们将描述我们如何使用Spark的强制方面来重新设计一个大型的、复杂的（100+阶段）管道，这个管道最初是用HQL over Hive编写的。特别是，我们将描述如何控制数据分布，避免数据倾斜，并实现特定于应用程序的优化，以构建性能可靠的数据管道。与之前的HQL查询集相比，这个新的基于Spark的管道是模块化的、可读的，并且更易于维护。除了质量上的改进，我们还观察到资源使用和数据登陆时间的减少。用例：N-gram语言模型训练自然语言处理是人工智能的一个领域，研究计算机与人类语言之间的交互作用。计算机可以训练成一种语言的模型，这些模型用于检测和纠正拼写错误。N元语言模型是目前应用最广泛的语言建模方法。一个N-gram通常被写成一个N个单词的短语，前N-1个单词作为历史，最后一个单词作为基于前N-1个单词的概率预测。例如，"Canyouplease come here"包含5个单词，是一个5克的单词。它的历史是"你能来吗？"基于这段历史，N-gram语言模型可以计算"here"这个词的条件概率大规模、高阶N-gram语言模型（例如N=5）在自动语音识别和机器翻译等许多应用中已被证明是非常有效的。例如，在Facebook，它被用来为上传到页面的视频自动生成字幕，并检测具有潜在低质量地名的页面（例如"Home sweet Home"，"Apt#00，Fake lane，Foo City"）。用大数据集训练的语言模型比用小数据集训练的语言模型具有更好的准确性。数据集越大，覆盖不常见单词（或N-gram）的大量实例的可能性就越大。对于较大数据集的训练，通常使用分布式计算框架（如MapReduce）来获得更好的可伸缩性和模型训练的并行化。早期解决方案我们最初开发了一个基于Hive的解决方案来生成N-gram语言模型。最后两个单词历史记录了N个G计数，使用基于C++的变换估计部分语言模型并将它们保存在蜂箱中。在不同的数据源上构建了不同的子模型，每个子模型都由配置单元查询触发。之后，所有这些子模型在最后一个作业中使用插值算法对每个子模型使用权重进行组合。以下是管道的高层次概述：基于Hive的解决方案在构建语言模型方面获得了合理的成功：当使用数十亿个N-gram进行训练时，它可以轻松地构建5-gram语言模型。当我们试图增加训练数据集的大小时，运行管道的端到端时间在不可接受的范围内。Hive提供了一个基于SQL的引擎，可以方便地编写查询，查询会自动转换为MapReduce作业。对于训练语言模型，将计算表示为SQL查询不是一种自然的适合，原因如下：管道代码包括每个子模型训练的几个SQL查询。这些查询基本上是相似的，只是略有不同。为模型训练编写新的管道会导致这些SQL查询的重复。随着越来越多的子句被添加到查询中，理解查询的意图变得越来越困难。更改查询的一部分需要重新运行整个管道，以确保它不会导致回归。无法单独测试更改会使开发周期花费更长的时间。另一种选择是，编写Hadoop作业在表达计算方面为开发人员提供了更多的自由，但是需要更多的时间，并且需要Hadoop方面的专业知识。基于火花的解决方案Spark提供了一种特定于域的语言（DSL），除了将作业作为SQL查询编写之外，它还可以方便地编写自定义应用程序。有了DSL，您可以控制较低级别的操作（例如，当数据被洗牌时）并可以访问中间数据。这有助于实现复杂算法，实现更高的效率和稳定性。它还允许用户以模块化的方式编写管道，而不是单一的SQL字符串，这提高了管道的可读性、可维护性和可测试性。这些优点使我们尝试使用Spark。重写C++语言——它在Scala或java中实现了语言模型训练算法——将是一个非常重要的努力，所以我们决定不改变这一部分。就像蜂箱一样，SCAK支持运行自定义用户代码，这使得调用相同的C++二进制文件变得容易。由于我们不必维护C++逻辑的两个版本，所以迁移是平滑的过渡，迁移对用户是透明的。我们没有使用sparksql，而是使用了Spark提供的RDD接口，因为它能够控制中间数据的分区并直接管理shard的生成。Spark的pipe（）运算符用于调用二进制文件。在更高的层次上，管道的设计保持不变。我们继续使用配置单元表作为应用程序的初始输入和最终输出。中间输出被写入集群节点上的本地磁盘。整个应用程序大约有1000行Scala代码，在Spark上执行时可以生成100多个stage（取决于训练数据源的数量）。可扩展性挑战当我们用更大的训练数据集测试Spark时，我们遇到了一些可伸缩性的挑战。在本节中，我们从描述数据分布需求（平滑和分片）开始，接着介绍它带来的挑战和我们的解决方案。平滑的N-gram模型是根据训练数据中的N-gram出现次数来估计的。由于训练数据中可能缺少N个gram，这可能很难将其推广到不可见的数据中。为了解决这个问题，人们使用了许多平滑方法，这些方法通常会对观测到的N-gram计数进行折扣，以提高看不见的N-gram概率，并使用低阶模型来平滑高阶模型。由于平滑，对于一个历史为h的N-图，需要对具有相同历史的所有N-图和以h为后缀的所有低阶N-图的计数来估计其概率。例如，对于三元组"how are you"，其中"how are"是历史，"you"是要预测的单词，为了估计P（you | how are），我们需要"how are*，""are*，"和所有的unigram（单个单词N-grams）的计数，其中*是表示词汇表中任何单词的通配符。频繁出现的N-gram（例如"how are*"）在处理时会导致数据倾斜。分片在分布式计算框架下，我们可以将N-gram计数划分为多个碎片，这样它们就可以由多台机器并行处理。基于N-gram历史的最后k个单词的切分可以保证所有比k个单词长的N-gram在碎片中得到适当的平衡。这需要在所有碎片中共享长度为k的所有N个克的计数。我们将所有这些短N-gram放入一个称为"0-shard"的特殊切分中，例如，如果k为2，那么从训练数据中提取的所有unigram和bigram被放在一个单独的shard（0-shard）中，并且所有进行模型训练的服务器都可以访问它们。问题：数据倾斜在基于Hive的流水线中，我们对两个词的历史切分进行模型训练。两个词历史切分意味着共享同一组最重要的两个词历史（最接近被预测的词）的所有N元计数被分配到同一个节点进行处理。与单词历史相比，两词切分通常会使数据分布更加均衡，只是所有节点都必须共享平滑算法所需的单元和双元组的统计信息，这些统计量存储在0-切分中。下图展示了一个单词和两个单词历史的碎片分布的比较。对于大型数据集，两个词的历史切分可以生成一个大的0分片。必须向所有节点分配一个大的0-shard会降低整个计算时间。它还可能导致潜在的作业不稳定，因为很难预先预测内存需求，而且一旦启动作业，它可能会在进程中耗尽内存。虽然可以预先分配更多内存，但它仍然不能保证100%的作业稳定性，并且可能导致整个集群内存利用率低下，因为并非所有作业实例都需要比历史平均值更多的内存。在尝试将计算转移到Spark时，作业可以以较小的工作负载运行。但在更大的数据集中，我们发现了以下问题：由于长时间没有收到执行器的心跳信号，驱动程序将executor标记为"lost"执行人OOM频繁执行者GC洗牌服务室块的火花限制为2GB在较高的层次上，所有这些问题的根本原因都可以归结为数据倾斜。我们希望实现碎片的均衡分布，但是两个单词的历史切分和单个单词的历史切分都没有帮助。所以我们想出了一个混合方法：渐进式