Hadoop中的Python框架的使用指南

更新時(shí)間：2015年04月22日 11:48:31 投稿：goldensun

這篇文章主要介紹了Hadoop中的Python框架的使用指南,Hadoop一般使用復(fù)雜的Java操作,但通過該框架使得Python腳本操作Hadoop成為了可能,需要的朋友可以參考下

最近，我加入了Cloudera，在這之前，我在計(jì)算生物學(xué)/基因組學(xué)上已經(jīng)工作了差不多10年。我的分析工作主要是利用Python語(yǔ)言和它很棒的科學(xué)計(jì)算棧來進(jìn)行的。但Apache Hadoop的生態(tài)系統(tǒng)大部分都是用Java來實(shí)現(xiàn)的，也是為Java準(zhǔn)備的，這讓我很惱火。所以，我的頭等大事變成了尋找一些Python可以用的Hadoop框架。

在這篇文章里，我會(huì)把我個(gè)人對(duì)這些框架的一些無關(guān)科學(xué)的看法寫下來，這些框架包括:

Hadoop流
mrjob
dumbo
hadoopy
pydoop
其它

最終，在我的看來，Hadoop的數(shù)據(jù)流(streaming)是最快也是最透明的選項(xiàng)，而且最適合于文本處理。mrjob最適合于在Amazon EMR上快速工作，但是會(huì)有顯著的性能損失。dumbo 對(duì)于大多數(shù)復(fù)雜的工作都很方便(對(duì)象作為鍵名(key))，但是仍然比數(shù)據(jù)流(streaming)要慢。

請(qǐng)繼續(xù)往下閱讀，以了解實(shí)現(xiàn)細(xì)節(jié)，性能以及功能的比較。

一個(gè)有趣的問題

為了測(cè)試不同的框架，我們不會(huì)做“統(tǒng)計(jì)詞數(shù)”的實(shí)驗(yàn)，轉(zhuǎn)而去轉(zhuǎn)化谷歌圖書N-元數(shù)據(jù)。 N-元代表一個(gè)n個(gè)詞構(gòu)成的元組。這個(gè)n-元數(shù)據(jù)集提供了谷歌圖書文集中以年份分組的所有1-，2-，3-，4-，5-元記錄的統(tǒng)計(jì)數(shù)目。在這個(gè)n-元數(shù)據(jù)集中的每行記錄都由三個(gè)域構(gòu)成：n-元，年份，觀測(cè)次數(shù)。(您能夠在http://books.google.com/ngrams取得數(shù)據(jù))。

我們希望去匯總數(shù)據(jù)以觀測(cè)統(tǒng)計(jì)任何一對(duì)相互臨近的詞組合所出現(xiàn)的次數(shù)，并以年份分組。實(shí)驗(yàn)結(jié)果將使我們能夠判斷出是否有詞組合在某一年中比正常情況出現(xiàn)的更為頻繁。如果統(tǒng)計(jì)時(shí)，有兩個(gè)詞在四個(gè)詞的距離內(nèi)出現(xiàn)過，那么我們定義兩個(gè)詞是“臨近”的。或等價(jià)地，如果兩個(gè)詞在2-，3-或者5-元記錄中出現(xiàn)過，那么我們也定義它們是”臨近“的。一次，實(shí)驗(yàn)的最終產(chǎn)物會(huì)包含一個(gè)2-元記錄，年份和統(tǒng)計(jì)次數(shù)。

有一個(gè)微妙的地方必須強(qiáng)調(diào)。n-元數(shù)據(jù)集中每個(gè)數(shù)據(jù)的值都是通過整個(gè)谷歌圖書語(yǔ)料庫(kù)來計(jì)算的。從原理上來說，給定一個(gè)5-元數(shù)據(jù)集，我可以通過簡(jiǎn)單地聚合正確的n-元來計(jì)算出4-元、3-元和2-元數(shù)據(jù)集。例如，當(dāng)5-元數(shù)據(jù)集包含

(the, cat, in, the, hat)    1999   20
(the, cat, is, on, youtube)  1999   13
(how, are, you, doing, today) 1986  5000

時(shí)，我們可以將它聚合為2-元數(shù)據(jù)集以得出如下記錄

(the, cat) 1999  33   // 也就是, 20 + 13

然而，實(shí)際應(yīng)用中，只有在整個(gè)語(yǔ)料庫(kù)中出現(xiàn)了40次以上的n元組才會(huì)被統(tǒng)計(jì)進(jìn)來。所以，如果某個(gè)5元組達(dá)不到40次的閾值，那么Google也提供組成這個(gè)5元組的2元組數(shù)據(jù)，這其中有一些或許能夠達(dá)到閾值。出于這個(gè)原因，我們用相鄰詞的二元數(shù)據(jù)，隔一個(gè)詞的三元組，隔兩個(gè)詞的四元組，以此類推。換句話說，與給定二元組相比，三元組多的只是最外層的詞。除了對(duì)可能的稀疏n元數(shù)據(jù)更敏感，只用n元組最外層的詞還有助于避免重復(fù)計(jì)算?？偟膩碚f，我們將在2元、3元、4元和5元數(shù)據(jù)集上進(jìn)行計(jì)算。

MapReduce的偽代碼來實(shí)現(xiàn)這個(gè)解決方案類似這樣：

def map(record):
  (ngram, year, count) = unpack(record)
  // 確保word1為字典第一個(gè)字
  (word1, word2) = sorted(ngram[first], ngram[last])
  key = (word1, word2, year)
  emit(key, count)
 
def reduce(key, values):
  emit(key, sum(values))

硬件

這些MapReduce組件在一個(gè)大約20GB的隨機(jī)數(shù)據(jù)子集上執(zhí)行。完整的數(shù)據(jù)集涵蓋1500個(gè)文件；我們用這個(gè)腳本選取一個(gè)隨機(jī)子集。文件名保持完整，這一點(diǎn)相當(dāng)重要，因?yàn)槲募_定了數(shù)據(jù)塊的n-元中n的值。

Hadoop集群包含5個(gè)使用CentOS 6.2 x64的虛擬節(jié)點(diǎn)，每個(gè)都有4個(gè)CPU，10GB RAM，100GB硬盤容量，并且運(yùn)行CDH4。集群每次能夠執(zhí)行20個(gè)并行運(yùn)算，每個(gè)組件能夠執(zhí)行10個(gè)減速器。

集群上運(yùn)行的軟件版本如下：

  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)庫(kù)中包含最新版本
  java:1.6

實(shí)現(xiàn)

大多數(shù)Python框架都封裝了Hadoop Streaming，還有一些封裝了Hadoop Pipes，也有些是基于自己的實(shí)現(xiàn)。下面我會(huì)分享一些我使用各種Python工具來寫Hadoop jobs的經(jīng)驗(yàn)，并會(huì)附上一份性能和特點(diǎn)的比較。我比較感興趣的特點(diǎn)是易于上手和運(yùn)行，我不會(huì)去優(yōu)化某個(gè)單獨(dú)的軟件的性能。

在處理每一個(gè)數(shù)據(jù)集的時(shí)候，都會(huì)有一些損壞的記錄。對(duì)于每一條記錄，我們要檢查是否有錯(cuò)并識(shí)別錯(cuò)誤的種類，包括缺少字段以及錯(cuò)誤的N元大小。對(duì)于后一種情況，我們必須知道記錄所在的文件名以便確定該有的N元大小。

所有代碼可以從 GitHub 獲得。

Hadoop Streaming

Hadoop Streaming 提供了使用其他可執(zhí)行程序來作為Hadoop的mapper或者reduce的方式，包括標(biāo)準(zhǔn)Unix工具和Python腳本。這個(gè)程序必須使用規(guī)定的語(yǔ)義從標(biāo)準(zhǔn)輸入讀取數(shù)據(jù)，然后將結(jié)果輸出到標(biāo)準(zhǔn)輸出。直接使用Streaming 的一個(gè)缺點(diǎn)是當(dāng)reduce的輸入是按key分組的時(shí)候，仍然是一行行迭代的，必須由用戶來辨識(shí)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)

下面是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)

Hadoop流(Streaming)默認(rèn)用一個(gè)tab字符分割健(key)和值(value)。因?yàn)槲覀円灿胻ab字符分割了各個(gè)域(field)，所以我們必須通過傳遞給Hadoop下面三個(gè)選項(xiàng)來告訴它我們數(shù)據(jù)的健(key)由前三個(gè)域構(gòu)成。

-jobconf stream.num.map.output.key.fields=3
-jobconf stream.num.reduce.output.key.fields=3

要執(zhí)行Hadoop任務(wù)命令

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

注意，mapper.py和reducer.py在命令中出現(xiàn)了兩次，第一次是告訴Hadoop要執(zhí)行著兩個(gè)文件，第二次是告訴Hadoop把這兩個(gè)文件分發(fā)給集群的所有節(jié)點(diǎn)。

Hadoop Streaming 的底層機(jī)制很簡(jiǎn)單清晰。與此相反，Python以一種不透明的方式執(zhí)行他們自己的序列化/反序列化，而這要消耗更多的資源。而且，如果Hadoop軟件已經(jīng)存在，Streaming就能運(yùn)行，而不需要再在上面配置其他的軟件。更不用說還能傳遞Unix 命令或者Java類名稱作 mappers/reducers了。

Streaming缺點(diǎn)是必須要手工操作。用戶必須自己決定如何將對(duì)象轉(zhuǎn)化為為成鍵值對(duì)（比如JSON 對(duì)象）。對(duì)于二進(jìn)制數(shù)據(jù)的支持也不好。而且如上面說過的，必須在reducer手工監(jiān)控key的邊界，這很容易出錯(cuò)。

mrjob

mrjob是一個(gè)開放源碼的Python框架，封裝Hadoop的數(shù)據(jù)流，并積極開發(fā)Yelp的。由于Yelp的運(yùn)作完全在亞馬遜網(wǎng)絡(luò)服務(wù)，mrjob的整合與EMR是令人難以置信的光滑和容易（使用 boto包）。

mrjob提供了一個(gè)Python的API與Hadoop的數(shù)據(jù)流，并允許用戶使用任何對(duì)象作為鍵和映射器。默認(rèn)情況下，這些對(duì)象被序列化為JSON對(duì)象的內(nèi)部，但也有支持pickle的對(duì)象。有沒有其他的二進(jìn)制I / O格式的開箱即用，但有一個(gè)機(jī)制來實(shí)現(xiàn)自定義序列化。

值得注意的是，mrjob似乎發(fā)展的非?？欤⒂泻芎玫奈臋n。

所有的Python框架，看起來像偽代碼實(shí)現(xiàn)：

#! /usr/bin/env python
 
import os
import re
 
from mrjob.job import MRJob
from mrjob.protocol import RawProtocol, ReprProtocol
 
class NgramNeighbors(MRJob):
 
  # mrjob allows you to specify input/intermediate/output serialization
  # default output protocol is JSON; here we set it to text
  OUTPUT_PROTOCOL = RawProtocol
 
  def mapper_init(self):
    # determine value of n in the current block of ngrams by parsing filename
    input_file = os.environ['map_input_file']
    self.expected_tokens = int(re.findall(r'([\d]+)gram', os.path.basename(input_file))[0])
 
  def mapper(self, key, line):
    data = line.split('\t')
 
    # error checking
    if len(data) < 3:
      return
 
    # unpack data
    ngram = data[0].split()
    year = data[1]
    count = int(data[2])
 
    # more error checking
    if len(ngram) != self.expected_tokens:
      return
 
    # generate key
    pair = sorted([ngram[0], ngram[self.expected_tokens - 1]])
    k = pair + [year]
 
    # note that the key is an object (a list in this case)
    # that mrjob will serialize as JSON text
    yield (k, count)
 
  def combiner(self, key, counts):
    # the combiner must be separate from the reducer because the input
    # and output must both be JSON
    yield (key, sum(counts))
 
  def reducer(self, key, counts):
    # the final output is encoded as text
    yield "%s\t%s\t%s" % tuple(key), str(sum(counts))
 
if __name__ == '__main__':
  # sets up a runner, based on command line options
  NgramNeighbors.run()

mrjob只需要安裝在客戶機(jī)上，其中在作業(yè)的時(shí)候提交。下面是要運(yùn)行的命令：

export HADOOP_HOME="/usr/lib/hadoop-0.20-mapreduce"
./ngrams.py -r hadoop --hadoop-bin /usr/bin/hadoop --jobconf mapred.reduce.tasks=10 -o hdfs:///output-mrjob hdfs:///ngrams

編寫MapReduce的工作是非常直觀和簡(jiǎn)單的。然而，有一個(gè)重大的內(nèi)部序列化計(jì)劃所產(chǎn)生的成本。最有可能的二進(jìn)制計(jì)劃將需要實(shí)現(xiàn)的用戶（例如，為了支持typedbytes）。也有一些內(nèi)置的實(shí)用程序日志文件的解析。最后，mrjob允許用戶寫多步驟的MapReduce的工作流程，在那里從一個(gè)MapReduce作業(yè)的中間輸出被自動(dòng)用作輸入到另一個(gè)MapReduce工作。

（注：其余的實(shí)現(xiàn)都非常相似，除了包具體的實(shí)現(xiàn)，他們都能被找到here.。）

dumbo

dumbo 是另外一個(gè)使用Hadoop流包裝的框架。dumbo出現(xiàn)的較早，本應(yīng)該被許多人使用，但由于缺少文檔，造成開發(fā)困難。這也是不如mcjob的一點(diǎn)。

dumbo通過typedbytes執(zhí)行序列化，能允許更簡(jiǎn)潔的數(shù)據(jù)傳輸，也可以更自然的通過指定JavaInputFormat讀取SequenceFiles或者其他格式的文件，比如，dumbo也可以執(zhí)行Python的egg和Java的JAR文件。

在我的印象中，我必須要手動(dòng)安裝dumbo中的每一個(gè)節(jié)點(diǎn)，它只有在typedbytes和dumbo以eggs形式創(chuàng)建的時(shí)候才能運(yùn)行。就像他會(huì)因?yàn)閛nMemoryErrors終止一樣，他也會(huì)因?yàn)槭褂媒M合器停止。

運(yùn)行dumbo任務(wù)的代碼是：

dumbo start ngrams.py \
    -hadoop /usr \
    -hadooplib /usr/lib/hadoop-0.20-mapreduce/contrib/streaming \
    -numreducetasks 10 \
    -input hdfs:///ngrams \
    -output hdfs:///output-dumbo \
    -outputformat text \
    -inputformat text

hadoopy

hadoopy 是另外一個(gè)兼容dumbo的Streaming封裝。同樣，它也使用typedbytes序列化數(shù)據(jù)，并直接把 typedbytes 數(shù)據(jù)寫到HDFS。

它有一個(gè)很棒的調(diào)試機(jī)制，在這種機(jī)制下它可以直接把消息寫到標(biāo)準(zhǔn)輸出而不會(huì)干擾Streaming過程。它和dumbo很相似，但文檔要好得多。文檔中還提供了與 Apache HBase整合的內(nèi)容。

用hadoopy的時(shí)候有兩種發(fā)發(fā)來啟動(dòng)jobs：

launch 需要每個(gè)節(jié)點(diǎn)都已經(jīng)安裝了Python/hadoopy ，但是在這之后的負(fù)載就小了。
launch_frozen 不要求節(jié)點(diǎn)上已經(jīng)安裝了Python，它會(huì)在運(yùn)行的時(shí)候安裝，但這會(huì)帶來15秒左右的額外時(shí)間消耗（據(jù)說通過某些優(yōu)化和緩存技巧能夠縮短這個(gè)時(shí)間）。

必須在Python程序中啟動(dòng)hadoopy job，它沒有內(nèi)置的命令行工具。

我寫了一個(gè)腳本通過launch_frozen的方式啟動(dòng)hadoopy

python launch_hadoopy.py

用launch_frozen運(yùn)行之后，我在每個(gè)節(jié)點(diǎn)上都安裝了hadoopy然后用launch方法又運(yùn)行了一遍，性能明顯好得多。

pydoop

與其他框架相比，pydoop 封裝了Hadoop的管道（Pipes），這是Hadoop的C++ API。正因?yàn)榇?，該?xiàng)目聲稱他們能夠提供更加豐富的Hadoop和HDFS接口，以及一樣好的性能。我沒有驗(yàn)證這個(gè)。但是，有一個(gè)好處是可以用Python實(shí)現(xiàn)一個(gè)Partitioner，RecordReader以及RecordWriter。所有的輸入輸出都必須是字符串。

最重要的是，我不能成功的從PIP或者源代碼構(gòu)建pydoop。

其他

happy 是一個(gè)用Jython來寫Hadoop job的框架，但是似乎已經(jīng)掛了
Disco 成熟的，非Hadoop 的 MapReduce.實(shí)現(xiàn)，它的核心使用Erlang寫的，提供了Python的API，它由諾基亞開發(fā)，不如Hadoop應(yīng)用廣泛。
octopy 是一個(gè)純Python的MapReduce實(shí)現(xiàn)，它只有一個(gè)源文件，并不適于“真正的”計(jì)算。
Mortar是另一個(gè)Python選擇，它不久前才發(fā)布，用戶可以通過一個(gè)網(wǎng)頁(yè)應(yīng)用提交Apache Pig 或者 Python jobs 處理放置在 Amazon S3上的數(shù)據(jù)。
有一些更高層次的Hadoop生態(tài)體系中的接口，像 Apache Hive和Pig。Pig 可以讓用戶用Python來寫自定義的功能，是通過Jython來運(yùn)行。 Hive 也有一個(gè)Python封裝叫做hipy。
(Added Jan. 7 2013) Luigi 是一個(gè)用于管理多步作業(yè)流程的Python框架。它與Apache Oozie 有一點(diǎn)相似，但是它內(nèi)置封裝了Hadoop Streaming(輕量級(jí)的封裝)。Luigi有一個(gè)非常好的功能是能夠在job出錯(cuò)的時(shí)候拋出Python代碼的錯(cuò)誤堆棧，而且它的命令行界面也非常棒。它的README文件內(nèi)容很多，但是卻缺少詳盡的參考文檔。Luigi 由Spotify 開發(fā)并在其內(nèi)部廣泛使用。

本地java

最后，我使用新的Hadoop Java API接口實(shí)施了MR任務(wù)，編譯完成后，這樣來運(yùn)行它：

hadoop jar /root/ngrams/native/target/NgramsComparison-0.0.1-SNAPSHOT.jar NgramsDriver hdfs:///ngrams hdfs:///output-native

關(guān)于計(jì)數(shù)器的特別說明

在我的MR jobs的最初實(shí)現(xiàn)里，我用計(jì)數(shù)器來跟蹤監(jiān)控不良記錄。在Streaming里，需要把信息寫到stderr。事實(shí)證明這會(huì)帶來不容忽視的額外開銷：Streaming job花的時(shí)間是原生java job的3.4倍。這個(gè)框架同樣有此問題。

將用Java實(shí)現(xiàn)的MapReduce job作為性能基準(zhǔn)。 Python框架的值是其相對(duì)于Java的性能指標(biāo)的比率。

2015422114844368.png (807×266)

Java明顯最快,，Streaming要多花一半時(shí)間，Python框架花的時(shí)間更多。從mrjob mapper的profile數(shù)據(jù)來看，它在序列化/反序列化上花費(fèi)了大量時(shí)間。dumbo和hadoopy在這方面要好一點(diǎn)。如果用了combiner 的話dumbo 還可以更快。

特點(diǎn)比較

大多來自各自軟件包中的文檔以及代碼庫(kù)。

2015422114912983.png (1190×483)