最近，我加入了Cloudera，在这之前，我在计算生物学/基因组学上已经工作了差不多10年。我的分析工作主要是利用Python语言和它很棒的科学计算栈来进行的。但Apache Hadoop的生态系统大部分都是用Java来实现的，也是为Java准备的，这让我很恼火。所以，我的头等大事变成了寻找一些Python可以用的Hadoop框架。
在这篇文章里，我会把我个人对这些框架的一些无关科学的看法写下来，这些框架包括:
- Hadoop流
- mrjob
- dumbo
- hadoopy
- pydoop
- 其它
最终，在我的看来，Hadoop的数据流(streaming)是最快也是最透明的选项，而且最适合于文本处理。mrjob最适合于在Amazon EMR上快速工作，但是会有显著的性能损失。dumbo 对于大多数复杂的工作都很方便(对象作为键名(key))，但是仍然比数据流(streaming)要慢。
请继续往下阅读，以了解实现细节，性能以及功能的比较。
一个有趣的问题
为了测试不同的框架，我们不会做“统计词数”的实验，转而去转化谷歌图书N-元数据。 N-元代表一个n个词构成的元组。这个n-元数据集提供了谷歌图书文集中以年份分组的所有1-，2-，3-，4-，5-元记录的统计数目。 在这个n-元数据集中的每行记录都由三个域构成：n-元，年份，观测次数。(您能够在http://books.google.com/ngrams取得数据)。
我们希望去汇总数据以观测统计任何一对相互临近的词组合所出现的次数，并以年份分组。实验结果将使我们能够判断出是否有词组合在某一年中比正常情况出现的更为频繁。如果统计时，有两个词在四个词的距离内出现过，那么我们定义两个词是“临近”的。 或等价地，如果两个词在2-，3-或者5-元记录中出现过，那么我们也定义它们是”临近“的。 一次，实验的最终产物会包含一个2-元记录，年份和统计次数。
有一个微妙的地方必须强调。n-元数据集中每个数据的值都是通过整个谷歌图书语料库来计算的。从原理上来说，给定一个5-元数据集，我可以通过简单地聚合正确的n-元来计算出4-元、3-元和2-元数据集。例如，当5-元数据集包含
(the, cat, in, the, hat) 1999 20 (the, cat, is, on, youtube) 1999 13 (how, are, you, doing, today) 1986 5000</div>
时，我们可以将它聚合为2-元数据集以得出如下记录
(the, cat) 1999 33 // 也就是, 20 + 13</div>
然而，实际应用中，只有在整个语料库中出现了40次以上的n元组才会被统计进来。所以，如果某个5元组达不到40次的阈值，那么Google也提供组成这个5元组的2元组数据，这其中有一些或许能够达到阈值。出于这个原因，我们用相邻词的二元数据，隔一个词的三元组，隔两个词的四元组，以此类推。换句话说，与给定二元组相比，三元组多的只是最外层的词。除了对可能的稀疏n元数据更敏感，只用n元组最外层的词还有助于避免重复计算。总的来说，我们将在2元、3元、4元和5元数据集上进行计算。
MapReduce的伪代码来实现这个解决方案类似这样：
def map(record): (ngram, year, count) = unpack(record) // 确保word1为字典第一个字 (word1, word2) = sorted(ngram[first], ngram[last]) key = (word1, word2, year) emit(key, count) def reduce(key, values): emit(key, sum(values))</div>
硬件
这些MapReduce组件在一个大约20GB的随机数据子集上执行。完整的数据集涵盖1500个文件；我们用这个脚本选取一个随机子集。文件名保持完整，这一点相当重要，因为文件名确定了数据块的n-元中n的值。
Hadoop集群包含5个使用CentOS 6.2 x64的虚拟节点，每个都有4个CPU，10GB RAM，100GB硬盘容量，并且运行CDH4。集群每次能够执行20个并行运算，每个组件能够执行10个减速器。
集群上运行的软件版本如下：
Hadoop:2.0.0-cdh4.1.2 Python:2.6.6 mrjob:0.4-dev dumbo:0.21.36 hadoopy:0.6.0 pydoop:0.7(PyPI)库中包含最新版本 java:1.6</div>
实现
大多数Python框架都封装了Hadoop Streaming，还有一些封装了Hadoop Pipes，也有些是基于自己的实现。下面我会分享一些我使用各种Python工具来写Hadoop jobs的经验，并会附上一份性能和特点的比较。我比较感兴趣的特点是易于上手和运行，我不会去优化某个单独的软件的性能。
在处理每一个数据集的时候，都会有一些损坏的记录。对于每一条记录，我们要检查是否有错并识别错误的种类，包括缺少字段以及错误的N元大小。对于后一种情况，我们必须知道记录所在的文件名以便确定该有的N元大小。
所有代码可以从 GitHub 获得。
Hadoop Streaming
Hadoop Streaming 提供了使用其他可执行程序来作为Hadoop的mapper或者reduce的方式，包括标准Unix工具和Python脚本。这个程序必须使用规定的语义从标准输入读取数据，然后将结果输出到标准输出。直接使用Streaming 的一个缺点是当reduce的输入是按key分组的时候，仍然是一行行迭代的，必须由用户来辨识key与key之间的界限。
下面是mapper的代码:
#! /usr/bin/env python
import os
import re
import sys
# determine value of n in the current block of ngrams by parsing the filename
input_file = os.environ['map_input_file']
expected_tokens = int(re.findall(r'([\d]+)gram', os.path.basename(input_file))[0])
for line in sys.stdin:
data = line.split('\t')
# perform some error checking
if len(data) < 3:
continue
# unpack data
ngram = data[0].split()
year = data[1]
count = data[2]
# more error checking
if len(ngram) != expected_tokens:
continue
# build key and emit
pair = sorted([ngram[0], ngram[expected_tokens - 1]])
print >>sys.stdout, "%s\t%s\t%s\t%s" % (pair[0], pair[1], year, count)
</div>
下面是reducer:
#! /usr/bin/env python
import sys
total = 0
prev_key = False
for line in sys.stdin:
data = line.split('\t')
curr_key = '\t'.join(data[:3])
count = int(data[3])
# found a boundary; emit current sum
if prev_key and curr_key != prev_key:
print >>sys.stdout, "%s\t%i" % (prev_key, total)
prev_key = curr_key
total = count
# same key; accumulate sum
else:
prev_key = curr_key
total += count
# emit last key
if prev_key:
print >>sys.stdout, "%s\t%i" % (prev_key, total)
</div>
Hadoop流(Streaming)默认用一个tab字符分割健(key)和值(value)。因为我们也用tab字符分割了各个域(field)，所以我们必须通过传递给Hadoop下面三个选项来告诉它我们数据的健(key)由前三个域构成。
-jobconf stream.num.map.output.key.fields=3 -jobconf stream.num.reduce.output.key.fields=3</div>
要执行Hadoop任务命令
hadoop jar /usr/lib/hadoop-0.20-mapreduce/contrib/streaming/hadoop-streaming-2.0.0-mr1-cdh4.1.2.jar \
-input /ngrams \
-output /output-streaming \
-mapper mapper.py \
-combiner reducer.py \
-reducer reducer.py \
-jobconf stream.num.map.output.key.fields=3 \
-jobconf stream.num.reduce.output.key.fields=3 \
-jobconf mapred.reduce.tasks=10 \
-file mapper.py \
-file reducer.py
</div>
注意，mapper.py和reducer.py在命令中出现了两次，第一次是告诉Hadoop要执行着两个文件，第二次是告诉Hadoop把这两个文件分发给集群的所有节点。
Hadoop Streaming 的底层机制很简单清晰。与此相反，Python以一种不透明的方式执行他们自己的序列化/反序列化，而这要消耗更多的资源。 而且，如果Hadoop软件已经存在