“工業(yè)流水線”的鼻祖，福特 T 型汽車[1]的電機裝配，將組裝過程拆成 29 道工序，將裝備時間由平均二十分鐘降到五分鐘，效率提升四倍 ，下圖圖源[2]。

T 型汽車裝配流水線

這種流水線的思想在數(shù)據(jù)處理過程中也隨處可見。其核心概念是：

1. 標(biāo)準(zhǔn)化的數(shù)據(jù)集合：對應(yīng)待組裝對象，是對數(shù)據(jù)處理中各個環(huán)節(jié)輸入輸出的一種一致性抽象。所謂一致，就是一個任意處理環(huán)節(jié)的輸出，都可以作為任意處理環(huán)節(jié)的輸入。
2. 可組合的數(shù)據(jù)變換：對應(yīng)單道組裝工序，定義了對數(shù)據(jù)進行變換的一個原子操作。通過組合各種原子操作，可以具有強大的表達力。

則，數(shù)據(jù)處理的本質(zhì)是：針對不同需求，讀取并標(biāo)準(zhǔn)化數(shù)據(jù)集后，施加不同的變換組合。

Unix 管道

Unix 管道是一項非常偉大的發(fā)明，體現(xiàn)了 Unix 的一貫哲學(xué)：

程序應(yīng)該只關(guān)注一個目標(biāo)，并盡可能把它做好。讓程序能夠互相協(xié)同工作。應(yīng)該讓程序處理文本數(shù)據(jù)流，因為這是一個通用的接口。

— Unix Pipe 機制發(fā)明者 Malcolm Douglas McIlroy

上述三句話哲學(xué)正體現(xiàn)了我們提到的兩點，標(biāo)準(zhǔn)化的數(shù)據(jù)集合——來自標(biāo)準(zhǔn)輸入輸出的文本數(shù)據(jù)流，可組合的數(shù)據(jù)變換——能夠協(xié)同工作的程序（如像 sort, head, tail 這種 Unix 自帶的工具，和用戶自己編寫的符合管道要求的程序）。

讓我們來看一個使用 Unix tools 和管道來解決實際問題的例子。假設(shè)我們有一些關(guān)于服務(wù)訪問的日志文件（var/log/nginx/access.log ，例子來自 DDIA[3] 第十章），日志的每一行格式如下：

// $remote_addr - $remote_user [$time_local] "$request"
// $status $body_bytes_sent "$http_referer" "$http_user_agent"
216.58.210.78 - - [27/Feb/2015:17:55:11 +0000] "GET /css/typography.css HTTP/1.1" 
200 3377 "http://martin.kleppmann.com/" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_5) 
AppleWebKit/537.36 (KHTML, like Gecko) Chrome/40.0.2214.115 Safari/537.36"

我們的需求是，統(tǒng)計出日志文件中最受歡迎的五個網(wǎng)頁。使用 Unix Shell ，我們會寫出類似的命令：

cat /var/log/nginx/access.log | # 讀取文件，打入標(biāo)準(zhǔn)輸出
    awk '{print $7}' | # 取出每行按空格分割的第七個字段
    sort             | # 對每行按字面值進行排序
    uniq -c          | # 歸并重復(fù)行，并給出重復(fù)次數(shù)
    sort -r -n       | # 按重復(fù)次數(shù)降序進行排序
    head -n 5          # 輸出前五行

可以看出上述 Shell 命令有以下幾個特點：

1. 每個命令實現(xiàn)的功能都很簡單（高內(nèi)聚）
2. 所有命令通過管道進行組合（低耦合），當(dāng)然這也要求可組合的程序只面向標(biāo)準(zhǔn)輸入、標(biāo)準(zhǔn)輸出進行編程，無其他副作用（比如輸出到文件）
3. 輸入輸出面向文本而非二進制

此外，Unix 的管道的另一大優(yōu)點是——流式的處理數(shù)據(jù)。也即所有程序中間結(jié)果并非都計算完成之后，才送入下一個命令，而是邊算邊送，從而達到多個程序并行執(zhí)行的效果，這就是流水線的精髓了。

當(dāng)然，管道也有缺點——只能進行線性的流水線排布，這也限制了他的表達能力。

GFS 和 MapReduce

MapReduce 是谷歌 2004 年的論文 MapReduce: Simplified Data Processing on Large Clusters[4] 提出的，用以解決大規(guī)模集群、并行數(shù)據(jù)處理的一種算法。GFS 是與 MapReduce 配套使用的基于磁盤的分布式文件系統(tǒng)。

MapReduce 算法主要分為三個階段：

1. Map：在不同機器上并行的對每個數(shù)據(jù)分區(qū)執(zhí)行用戶定義的 map() → List<Key, Value> 函數(shù)。
2. Shuffle：將 map 的輸出結(jié)果（KV 對）按 key 進行重新分區(qū)，按 key 聚集送到不同機器上， Key→ List<Value>。
3. Reduce：在不同機器上并行地對 map 輸出的每個 key 對應(yīng)的List<Value> 調(diào)用 reduce 函數(shù)。

DDIA 第十章 MapReduce 執(zhí)行示意圖

每個 MapReduce 程序就是對存儲在 GFS 上的數(shù)據(jù)集（標(biāo)準(zhǔn)化的數(shù)據(jù)集）的一次變換。理論上，我們可以通過組合多個 MapReduce 程序（可組合的變換），來滿足任意復(fù)雜的數(shù)據(jù)處理需求。

但與管道不同的是，每次 MapReduce 的輸出都要進行“物化”，即完全落到分布式文件系統(tǒng) GFS 上，才會執(zhí)行下一個 MapReduce 程序。好處是可以進行任意的、非線性的 MapReduce 程序排布。壞處是代價非常高，尤其考慮到 GFS 上的文件是多機多副本的數(shù)據(jù)集，這意味著大量的跨機器數(shù)據(jù)傳輸、額外的數(shù)據(jù)拷貝開銷。

但要考慮到歷史上開創(chuàng)式的創(chuàng)新，縱然一開始缺點多多，但會隨著時間迭代而慢慢克服。GFS + MapReduce 正是這樣一種在工業(yè)界開創(chuàng)了在大規(guī)模集群尺度上處理海量數(shù)據(jù)的先河。

Spark

Spark 便是為了解決 MapReduce 中每次數(shù)據(jù)集都要落盤的一種演進。

首先，Spark 提出了標(biāo)準(zhǔn)的數(shù)據(jù)集抽象——RDD[5]，這是一種通過分片的形式分散在多機上、基于內(nèi)存的數(shù)據(jù)集?；趦?nèi)存可以使得每次處理結(jié)果不用落盤，從而處理延遲更低?；诜制梢允沟迷跈C器宕機時，只用恢復(fù)少量分片，而非整個數(shù)據(jù)集。邏輯上，我們可以將其當(dāng)做一個整體來進行變換，物理上，我們使用多機內(nèi)存承載其每個分片。

其次，基于 RDD，Spark 提供了多種可靈活組合的算子集，這相當(dāng)于對一些常用的變換邏輯進行“構(gòu)件化”，可以讓用戶開箱即用。（下面圖源 RDD 論文[6]）

RDD 論文中列出的算子

基于此，用戶可以進行任意復(fù)雜數(shù)據(jù)處理，在物理上多個數(shù)據(jù)集（點）和算子（邊）會構(gòu)成一個復(fù)雜的 DAG （有向無環(huán)圖）執(zhí)行拓?fù)洌?/p>

RDD 和算子構(gòu)成的 DAG

關(guān)系型數(shù)據(jù)庫

關(guān)系型數(shù)據(jù)庫是數(shù)據(jù)處理系統(tǒng)的集大成者。一方面，它對外提供強大的聲明式查詢語言——SQL，兼顧了靈活性和易用性。另一方面，他對內(nèi)使用緊湊、索引友好的存儲方式，可以支撐高效的數(shù)據(jù)查詢需求。關(guān)系型數(shù)據(jù)庫系統(tǒng)同時集計算和存儲于一身，又會充分利用硬盤，甚至網(wǎng)絡(luò)（分布式數(shù)據(jù)庫）特點，是對計算機各種資源全方位使用的一個典范。本文不去過分展開關(guān)系型數(shù)據(jù)庫實現(xiàn)的各個環(huán)節(jié)，而是聚焦本文重點——標(biāo)準(zhǔn)的數(shù)據(jù)集和可組合的算子。

關(guān)系型數(shù)據(jù)庫對用戶提供的數(shù)據(jù)基本組織單位是——關(guān)系，或者說表。在 SQL 模型中，這是一種由行列組成的、強模式的二維表。所謂強模式，可以在邏輯上理解為表格中每個單元所存儲的數(shù)據(jù)必須要符合該列“表頭”的類型定義。針對這種標(biāo)準(zhǔn)的二維表，用戶可以施加各種關(guān)系代數(shù)算子（選擇、投影、笛卡爾乘積）。

一條 SQL 語句，在進入 RDBMS 之后，經(jīng)過解析、校驗、優(yōu)化，最后轉(zhuǎn)化成算子樹進行執(zhí)行。對應(yīng)的 RDBMS 中的邏輯單元，我們通常稱之為——執(zhí)行引擎，F(xiàn)acebook Velox[7] 就是專門針對該生態(tài)位的一個 C++ 庫。

傳統(tǒng)的執(zhí)行引擎多使用火山模型，一種屬于拉（ pull-based ）流派的執(zhí)行方式。其基本概念就是以樹形的方式組織算子，并從根節(jié)點開始，自上而下的進行遞歸調(diào)用，算子間自下而上的以行（row）或者批（batch）的粒度返回數(shù)據(jù)。

近些年來，基于推（push-based）的流派漸漸火起來了，DuckDB、Velox 都屬于此流派。類似于將遞歸轉(zhuǎn)化為迭代，自下而上，從葉子節(jié)點進行計算，然后推給父親節(jié)點，直到根節(jié)點。每個算子樹都可以拆解為多個可以并行執(zhí)行的算子流水線（下圖源，F(xiàn)acebook Velox 文檔[8]）

我們把上圖順時針旋轉(zhuǎn)九十度，可以發(fā)現(xiàn)他和 Spark 的執(zhí)行方式如出一轍，更多關(guān)于 velox 機制的解析，可以參考我寫的這篇文章[9]。

但無論推還是拉，其對數(shù)據(jù)集和算子的抽象都符合本文一開始提出的理論。

小結(jié)

考察完上述四種系統(tǒng)之后，可以看出，數(shù)據(jù)處理在某種角度上是大一統(tǒng)的——首先抽象出歸一化的數(shù)據(jù)集，然后提供施加于該數(shù)據(jù)集之上的運算集，最終通過組合的形式表達用戶的各種數(shù)據(jù)處理需求。

參考資料

[1]福特 T 型汽車: https://www.youtube.com/watch?v=As0lqsd2-NI

[2]汽車流水線圖源: https://www.motor1.com/features/178264/ford-model-t-factory-cutaway-kimble/

[3]DDIA: https://ddia.qtmuniao.com/

[4]MapReduce 論文: https://research.google.com/archive/mapreduce-osdi04.pdf

[5]RDD 分析: https://www.qtmuniao.com/2019/11/14/rdd/

[6]RDD 論文: https://www.usenix.org/system/files/conference/nsdi12/nsdi12-final138.pdf

[7]Facebook Velox: https://github.com/facebookincubator/velox

[8]Facebook Velox 文檔: https://facebookincubator.github.io/velox/develop/task.html

[9]Facebook velox 運行機制解析: https://zhuanlan.zhihu.com/p/614918289