原文見(jiàn)：https://v-yop.fun/2022/03-18MapReduce%EF%BC%8CHive%EF%BC%8CSpark%E7%9A%84%E4%B8%A4%E4%B8%AA%E7%A4%BA%E4%BE%8B%E2%80%94%E2%80%94Word-Count%E5%92%8CJOIN.html

本文章采用 CC BY-NC-SA 4.0 協(xié)議 ，轉(zhuǎn)載請(qǐng)注明出處！

來(lái)進(jìn)行一些超有趣的事情——分別使用 MapReduce，Hive，Spark 來(lái)實(shí)現(xiàn)同一個(gè)程序，感受一下三者代碼的差異，這里選擇去實(shí)現(xiàn)一下 WordCount，以及一個(gè)表連接的示例。

Word Count

Word Count，不從它開(kāi)始從誰(shuí)開(kāi)始呢？首先是 MapReduce，使用 Java 語(yǔ)言，代碼是龐然大物。

可以看到，Mapper 和 Reducer 的代碼是很顯然的，在 Mapper 中，我們將每一行字符串按空格分割，并構(gòu)造`(WORD, 1)`的鍵值對(duì)，然后我們通過(guò) Combiner 進(jìn)行本地聚集，再發(fā)送給各個(gè) Reducer，每個(gè) Reducer 對(duì)每個(gè) KEY 對(duì)應(yīng)的所有值，進(jìn)行 sum 操作，得到結(jié)果。但這里卻需要定義一堆東西，更別說(shuō) Driver 里的許多東西實(shí)際上都是重復(fù)的。

我們?cè)賮?lái)看看 Scala 的 Spark 的代碼，Spark 實(shí)現(xiàn) WordCount 其實(shí)是對(duì)我們這些開(kāi)發(fā)人員最舒服的，但我還沒(méi)系統(tǒng)學(xué)習(xí)過(guò) Spark，所以不知道自己的描述是否正確。

不要看著 flatMap 就想入非非，RDD 不是 Monad！但雖然 RDD 不是 Monad，它仍舊可以使用 for，但這時(shí)它的上下文是列表的上下文——flatMap 函數(shù)的函數(shù)參數(shù)不能返回 RDD，因此我們?cè)?for 里所能做的只有列表能做的。

另外，我不知道為何 Spark 最后得到的結(jié)果的 KEY 為何是無(wú)序的……按理說(shuō)經(jīng)過(guò) shuffle，這里應(yīng)當(dāng)是有序的才對(duì)，我只能猜測(cè)，Spark 利用單進(jìn)程的方便之處，在折疊時(shí)是并行進(jìn)行的，并輸出到同一個(gè)文件中，放到 MapReduce 的語(yǔ)境下，就是多個(gè) Reducer 的輸出文件為同一個(gè)，這樣無(wú)論如何也不可能得出有序的結(jié)果。但這也是符合需求的——我確實(shí)沒(méi)有指定排序。

然后是 Hive，這里展示了 Hive 從建模，讀取數(shù)據(jù)到寫(xiě)出數(shù)據(jù)的全流程。

(SELECT explode(split(line, ' ')) AS word FROM docs)需要特別解釋一下，這里使用了所謂的表生成函數(shù)（UDTF） explode，即通過(guò)一行數(shù)據(jù)生成一個(gè)表（該函數(shù)處理一個(gè) Array，生成一個(gè)表，而 split 得到的是 Array），這里是 docs 表中每行數(shù)據(jù)按空格進(jìn)行切割，并將每個(gè)結(jié)果展平，最后得到一張表，表中每一個(gè)記錄都是一個(gè)單詞，可以認(rèn)為這是一種 flatMap，反過(guò)來(lái)說(shuō)，這種在 Mapper 階段干 flatMap 的操作，是顯然需要使用 UDTF 的。

在得到這個(gè)單詞表后，我們將其按照單詞進(jìn)行分組，并對(duì)每個(gè)分組進(jìn)行 COUNT 聚集，得到結(jié)果并用該查詢(xún)結(jié)果創(chuàng)建一張新表。通過(guò)SELECT *可以發(fā)現(xiàn)其生成的結(jié)果和 MapReduce 的版本一致。

JOIN

現(xiàn)在考慮一個(gè)傳統(tǒng)的案例——現(xiàn)在有一張部門(mén)表和一張雇員表，其中雇員屬于特定部門(mén)，且有自己的工資，現(xiàn)在要求獲取每個(gè)部門(mén)的工資最大的雇員的信息，相關(guān)表的定義如下。

這個(gè)需求和 SQL 非常契合，所以我們先使用 Hive 進(jìn)行描述，我們顯然需要一個(gè)內(nèi)連接來(lái)進(jìn)行此工作。（可能有更簡(jiǎn)單的方式，但我還沒(méi)學(xué)到）

這里使用了 RANK 窗口函數(shù)，其中根據(jù)部門(mén)分區(qū)，根據(jù)工資降序，獲取 rank 為 1 的就是工資最高的，這里耗時(shí) 28 秒。一個(gè)更加清晰的表述使用子查詢(xún)，但 Hive 不支持子查詢(xún)，所以這里使用 JOIN 來(lái)表述。

這個(gè)更加麻煩一些，而且性能更加差——兩次 JOIN，要執(zhí)行兩次 MapReduce 任務(wù)才行，它耗時(shí) 64 秒（Hive 不是會(huì)進(jìn)行 MapJoin 優(yōu)化嗎？？）。下面的 MapReduce 只描述第一種。

關(guān)于 MapReduce 的編寫(xiě)，考慮到部門(mén)和雇員是一對(duì)多的關(guān)系，且部門(mén)的數(shù)量一定是較少的，我們本可以使用 map 端 join，但為了通用性，我們使用 reduce 端 join，為此，用于 join 的鍵必須要在鍵值對(duì)的鍵中，因此，Mapper 的輸出的鍵中應(yīng)當(dāng)包含 deptno，然后在 reduce 的值的集合里，我們要讓部門(mén)作為第一個(gè)元素，因此我們需要為此創(chuàng)建一個(gè)虛擬鍵。

再考慮分區(qū)，分組，排序，關(guān)于分區(qū)，肯定是按 deptno 進(jìn)行分區(qū)；關(guān)于分組，我們也是按 deptno 進(jìn)行分組；關(guān)于排序，為了能保證部門(mén)在最前面，我們要按 deptno 和虛擬鍵進(jìn)行排序，這里也可以把工資也加進(jìn)來(lái)進(jìn)行排序，以保證第二個(gè)元素就能獲取到最高的工資，但這其實(shí)并無(wú)必要。

根據(jù)上面的分析，我們規(guī)定，Mapper 的輸出類(lèi)型為<(deptno, virtualKey), Text>（懶得定義一個(gè)泛類(lèi)型了，機(jī)械操作太多了），按 deptno 進(jìn)行分區(qū)，按 deptno 進(jìn)行分組，按 deptno、virtualKey 進(jìn)行排序；同時(shí)也可以發(fā)現(xiàn)，這里可以有一個(gè) Combiner；Reducer 的輸出類(lèi)型就隨意了，人能看就行。

MapReduce 編寫(xiě)這種業(yè)務(wù)代碼真的是非?！浅Ｍ纯?，底層細(xì)節(jié)太多了，遠(yuǎn)不如 SQL 那樣直接，更不用說(shuō)有三個(gè)以上的表進(jìn)行 reduce 端 join 的時(shí)候需要使用多個(gè) MR 任務(wù)，且需要注意的細(xì)節(jié)更多，需要定義更多 Bean……

然后是 Spark 版本。Spark 版本怎么寫(xiě)其實(shí)我還沒(méi)有學(xué)過(guò)，但跟隨著直覺(jué)，還是挺容易的。

Spark 的代碼真是優(yōu)雅！但一個(gè)遺憾的地方是函數(shù)參數(shù)沒(méi)法像 Haskell 那樣解構(gòu)，不太痛快！

總結(jié)

總結(jié)個(gè)啥，我只能說(shuō) Spark 太香了。上面的 JOIN 代碼，如果在工程實(shí)踐中應(yīng)當(dāng)定義 Bean，MapReduce 進(jìn)行連接操作時(shí)應(yīng)該使用 GenericWritable 作為 VALUE。實(shí)際上本還想寫(xiě)個(gè)分布式排序示例，但我對(duì) Hive 和 Spark 都還沒(méi)學(xué)到那里，先跳過(guò)了。