摘要：說(shuō)明本文所有操作均在環(huán)境下進(jìn)行。任何可以使用來(lái)編寫的應(yīng)用，最終會(huì)由編寫。書中分別介紹了如何使用和結(jié)合進(jìn)行開發(fā)。工具會(huì)創(chuàng)建作業(yè)，發(fā)送給各個(gè)，同時(shí)監(jiān)控整個(gè)作業(yè)的執(zhí)行過程。準(zhǔn)備好的運(yùn)行環(huán)境之后開始搭建的運(yùn)行環(huán)境，參考單節(jié)點(diǎn)集群配置。

本文所有操作均在 linux 環(huán)境下進(jìn)行。

轉(zhuǎn)載請(qǐng)注明出處。

"任何可以使用JavaScript來(lái)編寫的應(yīng)用，最終會(huì)由JavaScript編寫。"

作為一名小前端，我深受 Jeff Atwood 前輩的鼓舞。上面這條定律便是他提出來(lái)的。

最近在學(xué)習(xí) Hadoop ，權(quán)威指南 中介紹到了 Hadoop Streaming，說(shuō) Hadoop Streaming 是 Hadoop 提供的一個(gè)編程工具，它允許用戶使用任何可執(zhí)行文件或者腳本文件作為 Mapper 和 Reducer 。書中分別介紹了如何使用 Ruby 和 Python 結(jié)合 Hadoop Streaming 進(jìn)行開發(fā)。沒有 JS，不開心。我們 JS 這么強(qiáng)大，一定也可以。。。

我們先來(lái)分析 Hadoop Streaming 的原理，如下：
mapper 和 reducer 會(huì)從標(biāo)準(zhǔn)輸入中讀取用戶數(shù)據(jù)，一行一行處理后發(fā)送給標(biāo)準(zhǔn)輸出。Streaming 工具會(huì)創(chuàng)建 MapReduce 作業(yè)，發(fā)送給各個(gè) TaskTracker，同時(shí)監(jiān)控整個(gè)作業(yè)的執(zhí)行過程。

分析完原理之后我們知道了只需構(gòu)造 mapper 和 reducer 即可，他們的工作是從標(biāo)準(zhǔn)輸入讀取用戶數(shù)據(jù)，以行（hang）為單位處理完成后發(fā)送到標(biāo)準(zhǔn)輸出。

JavaScript 如何從標(biāo)準(zhǔn)輸入輸出讀寫數(shù)據(jù)呢？別擔(dān)心，我們有 NodeJS。
準(zhǔn)備好 JavaScript 的運(yùn)行環(huán)境之后開始搭建 Hadoop 的運(yùn)行環(huán)境，參考 Hadoop: 單節(jié)點(diǎn)集群配置。

先貼目錄結(jié)構(gòu)：

$ find .
.
./map
./reduce
./wordcount.txt

map 中的代碼如下：

#!/usr/bin/env node

// 引入readline模塊
const readline = require("readline")

// 創(chuàng)建readline接口實(shí)例
const rl = readline.createInterface({
    input:process.stdin,
    output:process.stdout
})

rl.on("line", line => {
    // 分離每一行的單詞
    line.split(" ").map((word) => {
        // 將單詞以如下格式寫入標(biāo)準(zhǔn)輸出
        console.log(`${word}	1`)
    })
})

rl.on("close", () => {
    process.exit(0)
})

reduce 中的代碼如下：

#!/usr/bin/env node
const readline = require("readline")

const  rl = readline.createInterface({
    input:process.stdin,
    output:process.stdout,
    terminal: false
})

// 存儲(chǔ)鍵值對(duì) 
let words = new Map()

rl.on("line", line => {
    // 解構(gòu)賦值
    const [word, count] = line.split("	")
    // 如果 Map 中沒有該單詞，則將該單詞放入 Map ，即第一次添加
    if (!words.has(word)) {
        words.set(word, parseInt(count))
    } else {
        // 如果該單詞已存在，則將該單詞對(duì)應(yīng)的 count 加 1
        words.set(word, words.get(word) + 1)
    }
})

rl.on("close", () => {
    words.forEach((v, k) => {
        // 將統(tǒng)計(jì)結(jié)果寫入標(biāo)準(zhǔn)輸出
        console.log(`${k}	${v}`)
    })
    process.exit(0)
})

wordcount.txt 中的內(nèi)容如下：

JS Java
JS Python
JS Hadoop

目前 map 和 reduce 這兩個(gè)程序還無(wú)法運(yùn)行，需要加可執(zhí)行權(quán)限，方法如下：

$ chmod +x map reduce

現(xiàn)在可以在終端測(cè)試一下程序是否能正確執(zhí)行：

$ cat wordcount.txt | ./map | ./reduce
JS      3
Java    1
Python  1
Hadoop  1

可以看到，已經(jīng)正確統(tǒng)計(jì)出了詞頻。

接下來(lái)只需把作業(yè)提交給 Hadoop ，讓它去執(zhí)行就可以了。

此時(shí)要確保 Hadoop 正常運(yùn)行

在 HDFS 中創(chuàng)建目錄：

$ hdfs dfs -mkdir input

將待處理文件上傳至 HDFS：

$ hdfs dfs -put wordcount.txt input

此時(shí)可以通過 web 接口查看文件是否正確上傳：

http://localhost:50070/explor...
如下圖所示：

向 Hadoop 提交作業(yè)

$ hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-2.7.3.jar 
> -input input/wordcount.txt 
> -output output 
> -mapper map 
> -reducer reduce 

檢查計(jì)算結(jié)果：

$ hdfs dfs -cat output/*
Hadoop    1
JS        3
Java      1
Python    1

可以看到與之前的結(jié)果一致。

解釋一下 Hadoop Streaming 的幾個(gè)參數(shù)：

-input：輸入文件路徑

-output：輸出文件路徑

-mapper：用戶自己寫的 mapper 程序，可以是可執(zhí)行文件或者腳本

-reducer：用戶自己寫的 reducer 程序，可以是可執(zhí)行文件或者腳本

Hadoop?Streaming?編程
Node.js 命令行程序開發(fā)教程
Readline?|?Node.js v7.7.0 Documentation