python - Python 中不同的并行模块

Question

这些天我要处理非常大的日志数据(压缩为7z后700GB)，性能问题至关重要。考虑到我正在工作的环境(8 核)，我正在考虑利用并行编程来实现更好的性能。目前我正在使用内置的多处理库，性能有所提高，但我想要更好。我听说还有很多其他的 python 并行编程库，例如 pp。

所以我的问题是这些模块之间有什么区别？有比其他更好的吗？

Answer 1

首先，有几个问题:

• 压缩了 700GB，那么未压缩的量是多少？
• 有多少个文件？
• 您想对这些日志做什么？我们如何划分和征服？

我认为您应该考虑使用 MapReduce 来处理如此大量的数据。

出于示例任务的目的，我假设您有 800GB 的压缩广告服务器事件日志数据，并且您想要执行一些简单的操作，例如计算该数据集中唯一用户的数量。对于这种数据量和这种处理方式，多处理将会有所帮助，但使用 MapReduce 会更快:我会研究 EMR和 MrJob或Dumbo 。执行简单的处理工作(例如用户计数)将有助于验证过程并帮助您开始从映射器和化简器角度思考问题。处理更复杂的任务需要花费更多的时间，但我认为如果您要在任何实际时间内处理如此大量的数据，那么这种投资是非常值得的。

例如，计算唯一用户数将从映射器开始，该映射器简单地获取每一行广告服务器数据并发出用户 ID(cookieID、IP 地址，以及我们可以用来区分用户的任何内容)。您还将拥有一个缩减程序，它将这些用户 ID 作为输入并删除或计算重复项。

当然，一旦您决定尝试一下，仍有大量工作要做。准备数据(将大文件拆分或将小文件分组为 blob，以便有效分配工作、存储未压缩的数据或以 EMR 的 Hadoop 风格理解的压缩格式存储数据)、调整 hadoop 变量以使用可用资源和您的算法、上传数据到s3等

从好的方面来说，您实际上应该能够在几个小时内处理 800GB 的数据。

Python 中的简单 MapReduce 示例:

这是日志文件格式:

AuctionID\tUserID\tSiteID\tURL\tUserAgent\tTimestamp

这只是一个简单的制表符分隔值 (tsv) 文件。

因此，我们将编写一个简单的映射器来从 stdin 中读取这样的行，并将 UserID 写入 stdout。

import sys

def usercount_mapper(input):
    for line in input:
        line = line.strip()
        parts = line.split("\t")
        user_id = parts[1]
        print "%s\t%s"%(user_id, 1)

if __name__=="__main__":
    usercount_mapper(sys.stdin)

还有一个简单的reducer实现来计算唯一的userId:

import sys

user_ids = {}
def usercount_reducer(input):
    for line in input:
        line = line.strip()
        user_id, count = line.split("\t")
        try:
            count = int(count)
        except ValueError:
            continue
        current_count = user_ids.get(user_id, 0)
        user_ids[user_id] = current_count + count

    for user_id, count in user_ids.iteritems():
        print "%s\t%s"%(user_id, count)

if __name__=="__main__":
    usercount_reducer(sys.stdin)

您可以在单个数据 block 上运行它，以便在本地测试它，只需执行以下操作:

$ cat mydata.tsv | map.py | sort | reduce.py > result.tsv

mapreduce 框架(如果您使用 EMR，则为 hadoop)将负责运行多个映射和化简任务，并在将数据传递给 reducer 之前对来自映射器的数据进行排序。为了让reducers真正完成他们的工作，MR框架还将散列键值(选项卡中的第一个值将输出与映射器分开(在本例中为UserID))，并将具有相同散列的映射器分发到相同的reducer。这样，id为4的用户将始终转到reducer 1，id 5将转到reducer 2，依此类推。

如果你想自己构建一些东西，你可以直接查看 Disco (Disco 是 Python 和 Erlang，所以如果您对 java 过敏，它可能是一个不错的选择:-))或 Hadoop构建您自己的 MapReduce 基础设施，而不是使用 EMR。在 Hadoop/EMR 世界中，还有一些很酷的数据处理平台，例如 Hive (用于描述数据和映射缩减算法的类似 SQL 的环境)或 Pig (就像 grep 和 awk 一样)这可能比上面的脚本更适合您。

例如，在 Hive 中表达您的架构后，您可以编写以下查询来获取唯一用户(假设您之前定义了表 users):

SELECT DISTINCT users.user_id FROM users;