1. 什么是Hadoop pipes？

Hadoop pipes允許用戶使用C++語言進行MapReduce程序設計。它采用的主要方法是將應用邏輯相關的C++代碼放在單獨的進程中，然后通過Socket讓Java代碼與C++代碼通信。從很大程度上說，這種方法類似于Hadoop Streaming，不同之處是通信方式不同：一個是標準輸入輸出，另一個是socket。

org.apache.hadoop.mapred.pipes.Submitter包中有一個public static方法用于提交作業(yè),該方法將作業(yè)封裝成一個JobConf對象和一個main方法（接收一個應用程序，可選的配置文件，輸入目錄和輸出目錄等），main方法的CLI(Client Line Interface)如下：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33

bin/hadoop pipes \ [-input inputDir] \ #輸入數(shù)據(jù)目錄 [-output outputDir] \ #輸出數(shù)據(jù)目錄 [-jar applicationJarFile] \? #應用程序jar包 [-inputformat class] \ #Java版的InputFormat [-map class] \ #Java版的Mapper [-partitioner class] \#Java版的Partitioner [-reduce class] \#Java版的Reducer [-writer class] \ #Java版的 RecordWriter [-program program url] \? #C++可執(zhí)行程序 [-conf configuration file] \#xml配置文件 [-D property=value] \ #配置JobConf屬性 [-fs local|namenode:port] \#配置namenode [-jt local|jobtracker:port] \#配置jobtracker [-files comma separated list of files] \ #已經(jīng)上傳文件到HDFS中的文件，它們可以像在本地一樣打開 [-libjars comma separated list of jars] \#要添加到classpath 中的jar包 [-archives comma separated list of archives]#已經(jīng)上傳到HDFS中的jar文件，可以 在程序中直接使用

本文主要介紹了Hadoop pipes的設計原理，包括設計架構，設計細節(jié)等。

2. Hadoop pipes設計架構

用戶通過bin/hadoop pipes將作業(yè)提交到org.apache.hadoop.mapred.pipes中的Submmit類，它首先會進行作業(yè)參數(shù)配置（調(diào)用函數(shù)setupPipesJob），然后通過JobClient(conf).submitJob(conf)將作業(yè)提交到Hadoop集群中。

在函數(shù)setupPipesJob中，Java代碼會使用ServerScoket創(chuàng)建服務器對象，然后通過ProcessBuilder執(zhí)行C++binary， C++binary實際上是一個Socket client，它從Java server中接收key/value數(shù)據(jù)，經(jīng)過處理（map，partition或者reduce等）后，返還給Java server，并由Java Server將數(shù)據(jù)寫到HDFS或者磁盤。

3. Hadoop pipes設計細節(jié)

Hadoop pipes允許用戶用C++編寫五個基本組件：mapper，reducer，partitioner，combiner，recordReader，這五個組件可以是Java編寫的，也可以是C++編寫的，下面分別介紹這幾個函數(shù)的執(zhí)行過程。

（1） mapper

Pipes會根據(jù)用戶的配置定制InputFormat，如果用戶要使用Java的InputFormat（hadoop.pipes.java.recordreader=true），則Hadoop會使用戶輸入的InputFormat（默認為TextInputFormat）；如果用戶使用C++的InputFormat，則Pipes Java端的代碼會讀取每個InputSplit，并調(diào)用downlink.runMap(reporter.getInputSplit(), job.getNumReduceTasks(), isJavaInput);通過socket傳輸給C++端的runMap(string _inputSplit, int _numReduces, bool pipedInput)函數(shù)。

在C++端，RecordReader會解析整個InputSplit，獲取數(shù)據(jù)來源（主要是文件路徑）和每個key/value對，并交給map函數(shù)處理，map將每個key/value的處理結果通過emit(const string& key, const string& value)函數(shù)返還給Java Server。

（2） paritioner

C++端處理完的結果會通過emit(const string& key, const string& value)函數(shù)傳給Java Server，以便將數(shù)據(jù)寫到磁盤上。在emit函數(shù)中，如果用戶定義了自己的paritioner，則Pipes會通過該函數(shù)判斷當前key/value將給哪個reduce task處理，并調(diào)用partitionedOutput(int reduce, const string& key,const string& value)函數(shù)將key/value傳遞給相應的reduce task。

（3） reducer

reducer的執(zhí)行過程與mapper基本一致。

4. 總結

Hadoop pipes給C++程序員提供了一個編寫MapReduce作業(yè)的方案，它使用socket讓Java和C++之間進行通信，這類似于thrift RPC的原理，也許Hadoop Pipes用thrift編寫會更加簡單。

Hadoop pipes使用Java代碼從HDFS上讀寫數(shù)據(jù)，并將處理邏輯封裝到C++中，數(shù)據(jù)會通過socket從Java傳輸給C++，這雖然增加了數(shù)據(jù)傳輸?shù)拇鷥r，但對于計算密集型的作業(yè)，其性能也許會有改進。

5. 參考資料

http://wiki.apache.org/hadoop/HowToDebugMapReducePrograms

http://cs.smith.edu/dftwiki/index.php/Hadoop_Tutorial_2.2_–_Running_C%2B%2B_Programs_on_Hadoop

http://www.itberry.com/?p=42

原創(chuàng)文章，轉載請注明：?轉載自董的博客

本文鏈接地址:?http://dongxicheng.org/mapreduce/hadoop-pipes-architecture/

總結

以上是生活随笔為你收集整理的Hadoop pipes设计原理的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。