我应该为这项任务学习/使用 MapReduce 还是其他类型的并行化？

Question

在与 Google 的一位朋友交谈后，我想实现某种 Job/Worker 模型来更新我的数据集。

此数据集反映了 3rd 方服务的数据，因此，要进行更新，我需要对其 API 进行多次远程调用。我认为将花费大量时间等待来自此第 3 方服务的响应。我想加快速度，并更好地利用我的计算时间，通过并行化这些请求并同时保持其中许多请求处于打开状态，因为它们等待各自的响应。

在我解释我的具体数据集并解决问题之前，我想澄清一下我正在寻找什么答案：

1. 这是一个非常适合与 MapReduce 并行化的流程吗？
2. 如果是，在 Amazon 的 mapreduce 模块上运行是否具有成本效益，该模块按小时计费，并在作业完成时向上取整？ （我不确定究竟什么才是“工作”，所以我不知道具体是如何计费的）
3. 如果否，我应该使用其他系统/模式吗？ 和是否有库可以帮助我在 python 中执行此操作（在 AWS 上，使用 EC2 + EBS）？
4. 您认为我设计此工作流程的方式有什么问题吗？

好的，现在进入细节：

数据集由拥有最喜欢项目和关注其他用户的用户组成。目的是能够更新每个用户的队列——用户在加载页面时将看到的项目列表，基于她关注的用户最喜欢的项目。但是，在我处理数据和更新用户队列之前，我需要确保我拥有最新的数据，这是 API 调用的来源。

我可以拨打两个电话：

• Get Followed Users -- 返回被请求用户关注的所有用户，以及
• Get Favorite Items -- 返回所请求用户的所有最喜欢的项目。

在我为正在更新的用户调用获取关注的用户之后，我需要为每个被关注的用户更新最喜欢的项目。只有当所有被关注用户的所有收藏都返回时，我才能开始处理该原始用户的队列。这个流程看起来像：

此流程中的工作包括：

• 为用户启动更新队列 -- 通过获取用户，然后更新用户，存储他们，然后为每个用户创建 Get Favorites 作业来启动该过程。
• 为用户获取收藏夹 -- 从 3rd 方服务请求并存储指定用户的收藏夹列表。
• 为用户计算新队列 -- 处理一个新队列，现在所有数据都已获取，然后将结果存储在应用程序层使用的缓存中。

所以，我的问题是：

1. 这是一个非常适合与 MapReduce 并行化的流程吗？我不知道它是否能让我启动 UserX 的流程，获取所有相关数据，然后只有在这一切都完成后才返回处理 UserX 的队列。
2. 如果是，在 Amazon 的 mapreduce 模块上运行是否具有成本效益，该模块按小时计费，并在作业完成时向上取整？如果我使用他们的模块，我可以等待开放 API 请求的“线程”数量是否有限制？
3. 如果否，我应该使用其他系统/模式吗？ 和是否有库可以帮助我在 python 中执行此操作（在 AWS 上，使用 EC2 + EBS？）？
4. 您认为我设计此工作流程的方式有什么问题吗？

感谢阅读，期待与大家讨论。

编辑，回应 JimR：

感谢您的中肯答复。自从我写了原始问题以来，在我的阅读中，我已经远离使用 MapReduce。我还没有确定我想如何构建它，但是当我真的只是想并行化 HTTP 请求时，我开始觉得 MapReduce 更适合分配/并行化计算负载。

我的“减少”任务，即获取所有获取的数据并将其处理成结果的部分，并不是计算密集型的。我很确定它最终会成为一个大型 SQL 查询，每个用户执行一两秒钟。

所以，我倾向于：

• 一个非 MapReduce Job/Worker 模型，用 Python 编写。我的一个 Google 朋友让我为此学习 Python，因为它的开销低且可扩展性好。
• 使用 Amazon EC2 作为计算层。我认为这意味着我还需要一个 EBS 切片来存储我的数据库。
• 可能使用亚马逊的简单消息队列。听起来这第 3 个亚马逊小部件旨在跟踪作业队列、将结果从一个任务移动到另一个任务的输入并优雅地处理失败的任务。它非常便宜。可能值得实施而不是自定义作业队列系统。

Answer 1

您描述的工作可能非常适合队列，或者队列和作业服务器的组合。它当然也可以作为一组 MapReduce 步骤工作。

对于工作服务器，我建议查看 Gearman。文档不是很好，但演示文稿很好地记录了它，Python 模块也相当不言自明。

基本上，您在作业服务器中创建函数，然后客户端通过 API 调用这些函数。这些函数可以同步或异步调用。在您的示例中，您可能希望异步添加“开始更新”作业。这将执行任何准备任务，然后异步调用“获取关注的用户”作业。该作业将获取用户，然后调用“更新关注的用户”作业。将所有的“获取UserA的收藏”和好友工作一次性提交，并同步等待所有的结果。当它们全部返回时，它将调用“计算新队列”作业。

这种仅作业服务器的方法最初会不太健壮，因为确保您正确处理错误和任何关闭的服务器和持久性会很有趣。

对于队列，SQS 是一个显而易见的选择。它坚如磐石，可以从 EC2 快速访问，而且价格便宜。并且在您刚开始时比其他队列更容易设置和维护。

基本上，您会将消息放入队列，就像您将作业提交到上面的作业服务器一样，只是您可能不会同步执行任何操作。不是同步进行“Get Favorites For UserA”之类的调用，而是异步调用它们，然后有一条消息说检查它们是否都完成了。您需要某种持久性（您熟悉的 SQL 数据库，或者如果您想完全使用 AWS，则需要 Amazon 的 SimpleDB）来跟踪工作是否完成 - 您无法在 SQS 中检查工作的进度（尽管您可以在其他队列中）。检查它们是否全部完成的消息将进行检查 - 如果它们没有全部完成，则不做任何事情，然后将在几分钟内重试该消息（基于 visibility_timeout）。否则，您可以将下一条消息放入队列中。

这种仅队列的方法应该是健壮的，假设您不会在不执行工作的情况下错误地使用队列消息。使用 SQS 很难犯这样的错误 - 你真的必须尝试。不要使用自动消费队列或协议 - 如果您出错，您可能无法确保将替换消息放回队列中。

在这种情况下，队列和作业服务器的组合可能很有用。您可以避免没有持久性存储来检查作业进度 - 作业服务器将允许您跟踪作业进度。您的“获取用户收藏”消息可以将所有“获取用户A/B/C 的收藏”作业放入作业服务器。然后，在队列中放置一条“检查所有已完成的收藏夹”消息，其中包含需要完成的任务列表（以及足够的信息来重新启动任何神秘消失的作业）。

积分：

作为 MapReduce 执行此操作应该相当容易。

您的第一份工作的输入将是您所有用户的列表。该地图将获取每个用户，获取关注的用户，并为每个用户及其关注的用户输出行：

"UserX" "UserA"
"UserX" "UserB"
"UserX" "UserC"

身份减少步骤将保持不变。这将形成第二个作业的输入。第二个作业的地图将获取每行的收藏夹（您可能希望使用 memcached 来防止通过 API 获取 UserX/UserA 组合和 UserY/UserA 的收藏夹），并为每个收藏夹输出一行：

"UserX" "UserA" "Favourite1"
"UserX" "UserA" "Favourite2"
"UserX" "UserA" "Favourite3"
"UserX" "UserB" "Favourite4"

此作业的 reduce 步骤会将其转换为：

 "UserX" [("UserA", "Favourite1"), ("UserA", "Favourite2"), ("UserA", "Favourite3"), ("UserB", "Favourite4")]

此时，您可能有另一个 MapReduce 作业来使用这些值为每个用户更新您的数据库，或者您可以使用一些与 Hadoop 相关的工具（例如 Pig、Hive 和 HBase）来管理您的数据库你。

我建议使用 Cloudera 的 Distribution for Hadoop 的 ec2 管理命令来创建和拆除 EC2 上的 Hadoop 集群（他们的 AMI 上设置了 Python），并使用 Dumbo 之类的工具（在 PyPI 上）来创建 MapReduce 作业，因为它允许您在本地/开发机器上测试您的 MapReduce 作业，而无需访问 Hadoop。

祝你好运！

Answer 2

似乎我们将使用Node.js 和Seq 流控制库。从我的流程地图/流程图转移到代码的存根非常容易，现在只需填写代码以连接到正确的 API。

感谢您的回答，他们在寻找我正在寻找的解决方案方面提供了很多帮助。

Answer 3

我正在处理需要解决的类似问题。我还在研究 MapReduce 并使用 Amazon 的 Elastic MapReduce 服务。

我非常相信 MapReduce 可以解决这个问题。实现是我被挂断的地方，因为我不确定我的减速器是否需要做任何事情。

我会在了解您（和我）的问题后回答您的问题，希望对您有所帮助。

1. 是的，我认为它会很合适。您可以考虑利用 Elastic MapReduce 服务的多步骤选项。您可以使用 1 Step 来获取用户正在关注的人，并使用另一个 step 为每个关注者编译一个曲目列表，并且第二步的 reducer 可能是构建缓存的那个。

2. 取决于您的数据集有多大以及运行它的频率。如果不知道数据集有多大（或将得到多大），很难说它是否具有成本效益。最初，它可能会非常划算，因为您不必管理自己的 hadoop 集群，也不必为 EC2 实例（假设是您使用的实例）支付费用来一直运行。一旦达到了长时间处理这些数据的地步，使用 Amazon 的 MapReduce 服务的意义可能会越来越小，因为您将始终有节点在线。

工作基本上就是您的 MapReduce 任务。它可以由多个步骤组成（每个 MapReduce 任务都是一个步骤）。处理完您的数据并完成所有步骤后，您的工作就完成了。因此，您实际上是在为 Hadoop 集群中的每个节点支付 CPU 时间。因此，T*n 其中 T 是处理数据所需的时间（以小时为单位），n 是您告诉亚马逊启动的节点数。

我希望这会有所帮助，祝你好运。我想听听您最终是如何实现 Mappers 和 Reducers 的，因为我正在解决一个非常相似的问题，但我不确定我的方法是否真的是最好的。