1. 什么是Hadoop pipes？

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

org.apache.hadoop.mapred.pipes.Submitter包中有一個public static方法用于提交作業,該方法將作業封裝成一個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] \ #輸入數據目錄 [-output outputDir] \ #輸出數據目錄 [-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++可執行程序 [-conf configuration file] \#xml配置文件 [-D property=value] \ #配置JobConf屬性 [-fs local|namenode:port] \#配置namenode [-jt local|jobtracker:port] \#配置jobtracker [-files comma separated list of files] \ #已經上傳文件到HDFS中的文件，它們可以像在本地一樣打開 [-libjars comma separated list of jars] \#要添加到classpath 中的jar包 [-archives comma separated list of archives]#已經上傳到HDFS中的jar文件，可以 在程序中直接使用

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

2. Hadoop pipes設計架構

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

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

3. Hadoop pipes設計細節

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

（1） mapper

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

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

（2） paritioner

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

（3） reducer

reducer的執行過程與mapper基本一致。

4. 總結

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

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

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

原創文章，轉載請注明：?轉載自董的博客

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

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。