[TOC]
到目前為止，我只講述了系統之間拷貝數據的理想機制。但是在存儲系統之間搬運字節不是所要講述內容的全部。 最終會發現，『日志』是流的另一種說法， 并且日志是[流處理](http://highlyscalable.wordpress.com/2013/08/20/in-stream-big-data-processing/)的核心。
但是，等會兒，流處理到底是什么呢？
如果你是上世紀90年代晚期或者21世紀初[數據庫](http://cs.brown.edu/research/aurora/vldb03_journal.pdf)?[文化](http://db.cs.berkeley.edu/papers/cidr03-tcq.pdf)或者成功了一半的[數據](http://www-03.ibm.com/software/products/us/en/infosphere-streams)?[基礎設施](http://en.wikipedia.org/wiki/StreamBase_Systems)?[產品](http://en.wikipedia.org/wiki/Truviso)的愛好者，那么你就可能會把流處理與建創`SQL`引擎或者『箱子和箭頭』（`boxes and arrows`）接口用于事件驅動的處理聯系起來。
如果你關注大量出現的開源數據庫系統，你就可能把流處理和一些這領域的系統關聯起來， 比如[`Storm`](http://storm-project.net/)、[`Akka`](http://akka.io/)、[`S4`](http://incubator.apache.org/s4)和[`Samza`](http://engineering.linkedin.com/distributed-systems/log-what-every-software-engineer-should-know-about-real-time-datas-unifying)。 但是大部分人會把這些系統看為異步消息處理系統，與支持群集的遠程過程調用（`RPC`）層沒什么差別 （而事實上這一領域一些系統確實是如此）。
這些觀點都有一些局限性。流處理即與`SQL`無關，也不局限于實時流處理。 還沒有根本的原因，限制你不能使用多種不同的語言來表達計算，處理昨天的或者一個月之前的流數據。
[](https://github.com/oldratlee/translations/blob/master/log-what-every-software-engineer-should-know-about-real-time-datas-unifying/images/19202244_GPDx.jpg)
我把流處理視為更廣泛的概念：持續數據流處理的基礎設施。 我認為計算模型可以像`MapReduce`或者分布式處理框架一樣通用，但是有能力生成低時延的結果。
處理模型的真正驅動力是數據收集方法。成批收集數據自然是批處理。當數據是持續收集的，自然也應該持續不斷地處理。
美國的統計調查是一個成批收集數據的經典例子。 統計調查周期性的開展，用的是蠻力調查，通過挨門挨戶的走訪統計美國公民的信息。 在1790年統計調查剛剛開始，這樣做是很合理的。 那時的數據收集本質就是面向批處理的，包括了騎馬到周邊人家，用紙筆記錄，然后把成批的記錄運輸到人們統計數據的中心站點。 現在，在描述這個統計過程時，人們立即會想到為什么我們不保留出生和死亡的記錄，這樣就可以算出人口統計信息，這些信息或是持續即時計算出來或者按需要時間隔計算。
這是一個極端的例子，但是現在大量的數據傳輸處理仍然依賴于周期性的轉錄和批量的傳輸和集成。 處理批量轉錄數據的唯一方法就是批量的處理。 但是隨著這些批處理被持續的數據輸入所取代，人們自然而然的開始向持續處理轉變，以平滑地使用所需的處理資源并且減少延遲。
例如在`LinkedIn`幾乎完全沒有批量數據收集。我們大部分的數據要么是活動數據或者要么是數據庫變更，兩者都是不間斷地發生的。 事實上，你想到的任何商業業務，底層的機制幾乎都是不間斷的處理，正如*Jack Bauer*所說的，事件的發生是實時的。 當數據以成批的方式收集，幾乎總是由這些原因所致：有一些人為的步驟；缺少數字化；或是非數字化流程的歷史古董不能自動化。 當使用郵件或者人工方式，傳輸和處理數據是非常緩慢的。剛開始轉成自動化時，總是保持著原來流程的形式，所以這樣的情況會持續相當長的時間。
每天運行的『批量』處理作業常常在模擬一種窗口大小是一天的持續計算。 當然，底層的數據其實總是在變化著的。 在`LinkedIn`，這樣的做法如此之常見（并且在`Hadoop`做到這些的實現機制如此之復雜）， 以至于我們實現了一整套[框架](http://engineering.linkedin.com/datafu/datafus-hourglass-incremental-data-processing-hadoop)來管理增量的`Hadoop`工作流。
由此看來，對于流處理我很容易得出不同觀點： 它處理的是包含時間概念的底層數據并且不需要靜態的數據快照， 所以可以以用戶可控頻率生產輸出而不是等待數據集的『都』到達后再生產輸出（譯注：數據是會持續的，所以實際上不會有『都』達到的時間點）。 從這個角度上講，流處理是廣義上的批處理，隨著實時數據的流行，流處理會是很重要處理方式。
那么，為什么流處理的傳統觀點大家之前會認為更合適呢？ 我個人認為最大的原因是缺少實時數據收集，使得持續處理之前是學術性的概念。
我覺得，是否缺少實時數據的收集決定了商用流處理系統的命運。 當他們的客戶還是用面向文件的每日批量處理完成`ETL`和數據集成時， 建設流處理系統的公司專注于提供處理引擎來連接實時數據流，而結果是當時幾乎沒有人真地有實時數據流。 其實我在`LinkedIn`工作的初期，有一家公司想把一個非常棒的流處理系統賣給我們， 但是因為當時我們的所有數據都按小時收集在的文件里， 所以用上這個系統我們能做到的最好效果就是在每小時的最后把這些文件輸入到流處理系統中。 他們意識到這是個普遍問題。 下面的這個異常案例實際上是證明上面規律： 流處理獲得一些成功的一個領域 —— 金融領域，這個領域在過去，實時數據流就已經標準化，并且流處理已經成為了瓶頸。
甚至于在一個健康的批處理的生態中，我認為作為一種基礎設施風格，流處理的實際應用能力是相當廣闊的。 我認為它填補了實時數據請求/響應服務和離線批量處理之間的缺口。現代的互聯網公司，我覺得大約25%的代碼可以劃分到這個情況。
事實證明，日志解決了流處理中最關鍵的一些技術問題，后面我會進一步講述， 但解決的最大的問題是日志使得多個訂閱者可以獲得實時的數據輸入。 對技術細節感興趣的朋友，我們已經開源了[`Samza`](http://samza.apache.org/)， 它正是基于這些理念建設的一個流處理系統。 很多這方面的應用的更多技術細節我們在[此文檔](http://samza.apache.org/learn/documentation/latest/)中有詳細的描述。
## [](https://github.com/oldratlee/translations/blob/master/log-what-every-software-engineer-should-know-about-real-time-datas-unifying/part3-logs-and-real-time-stream-processing.md#數據流圖data-flow-graphs)數據流圖（`data flow graphs`）
[](https://github.com/oldratlee/translations/blob/master/log-what-every-software-engineer-should-know-about-real-time-datas-unifying/images/dag.png)
流處理最有趣的特點是它與流處理系統的內部組織無關， 但是與之密切相關的是，流處理是怎么擴展了之前在數據集成討論中提到的認識：輸入數據（`data feed`）是什么。 我們主要討論了原始數據（`primary data`）的`feeds`?或說 日志 —— 各種系統執行所產生的事件和數據行。 但是流處理允許我們包括了由其它`feeds`計算出的`feeds`。 在消費者看來，這些派生的`feeds`和 用于生成他們的原始數據的`feeds`?看下來沒什么差別。 這些派生的`feeds`可以按任意的復雜方式封裝組合。
讓我們再深入一點這個問題。 對于我們的目標，流處理作業是指從日志讀取數據和將輸出寫入到日志或其它系統的任何系統。 用于輸入和輸出的日志把這些處理系統連接成一個處理階段的圖。 事實上，以這樣的風格使用中心化的日志，你可以把組織全部的數據抓取、轉化和工作流僅僅看成是一系列的寫入它們的日志和處理過程。
流處理器根本不需要高大上的框架： 可以是讀寫日志的一個處理或者一組處理過程，但是為了便于管理處理所用的代碼，可以提供一些額外的基礎設施和支持。
在集成中日志的目標是雙重的：
首先，日志讓各個數據集可以有多個訂閱者并使之有序。 讓我們回顧一下『狀態復制』原理來記住順序的重要性。 為了更具體地說明，設想一下從數據庫中更新數據流 —— 如果在處理過程中把對同一記錄的兩次更新重新排序，可能會產生錯誤的輸出。 這里的有序的持久性要強于`TCP`之類所提供的有序，因為不局限于單一的點對點鏈接，并且在流程處理失敗和重連時仍然要保持有序。
其次，日志提供了處理流程的緩沖。 這是非常基礎重要的。如果多個處理之間是非同步的，那么生成上行流數據的作業生成數據可能比另一個下行流數據作業所能消費的更快。 這種情況下，要么使處理進程阻塞，要么引入緩沖區，要么丟棄數據。 丟棄數據似乎不是個好的選擇，而阻塞處理進程，會使得整個的數據流的圖被迫中止處理。 日志是一個非常非常大的緩沖，允許處理進程的重啟或是失敗，而不影響流處理圖中的其它部分的處理速度。 要擴展數據流到一個更龐大的組織，這種隔離性極其重要，整個處理是由組織中不同的團隊提供的處理作業完成的。 不能因為某個作業發生錯誤導致影響前面作業，結果整個處理流程都被卡住。
[`Storm`](http://storm-project.net/)和[`Sama`](http://samza.apache.org/)都是按這種風格構建，能用`kafka`或其它類似的系統作為它們的日志。
## [](https://github.com/oldratlee/translations/blob/master/log-what-every-software-engineer-should-know-about-real-time-datas-unifying/part3-logs-and-real-time-stream-processing.md#有狀態的實時流處理)有狀態的實時流處理
一些實時流處理做的只是無狀態的單次記錄的轉換，但有很多使用方式需要在流處理的某個大小的時間窗口內進行更復雜的計數、聚合和關聯（`join`）操作。 比如，給一個的事件流（如用戶點擊的流）附加上做點擊操作用戶的信息， —— 實際上即是關聯點擊流到用戶的賬戶數據庫。 這類流程最終總是要處理者維護一些狀態信息： 比如在計算一個計數時，需要維護到目前為止的計數器。 在處理者可能掛掉的情況下，如何維護正確的狀態？
最簡單的方案是把狀態保存在內存中。但是如果處理流程崩潰，會丟失中間狀態。 如果狀態是按窗口維護的，處理流程只能會回退到日志中窗口開始的時間點上。 但是，如果計數的時間窗口是1個小時這么長，那么這種方式可能不可行。
另一個方案是簡單地存儲所有的狀態到遠程的存儲系統，通過網絡與這些存儲關聯起來。 但問題是沒了數據的局部性并產生很多的網絡間數據往返（`network round-trip`）。
如何才能即支持像處理流程一樣分片又支持像『表』一樣的存儲呢？
回顧一下關于表和日志二象性的討論。它正好提供了把流轉換成與這里我們處理中所需的表的工具，同時也是一個解決表的容錯的處理機制。
流處理器可以把它的狀態保存在本地的『表』或『索引』中 ——?[`bdb`](http://www.oracle.com/technetwork/products/berkeleydb)、[`leveldb`](https://code.google.com/p/leveldb)?甚至是些更不常見的組件，如[`Lucene`](http://lucene.apache.org/)或[`fastbit`](https://sdm.lbl.gov/fastbit)索引。 這樣一些存儲的內容可以從它的輸入流生成（可能做過了各種轉換后的輸入流）。 通過記錄關于本地索引的變更日志，在發生崩潰、重啟時也可以恢復它的狀態。 這是個通用的機制，用于保持 任意索引類型的分片之間相互協作（`co-partitioned`）的本地狀態 與 輸入流數據 一致。
當處理流程失敗時，可以從變更日志中恢復它的索引。 每次備份時，即是日志把本地狀態轉化成一種增量記錄。
這種狀態管理方案的優雅之處在于處理器的狀態也是做為日志來維護。 我們可以把這個日志看成是數據庫表變更的日志。 事實上，這些處理器本身就很像是自維護的分片之間相互協作的表。 因為這些狀態本身就是日志，所以其它處理器可以訂閱它。 如果處理流程的目標是更新結點的最后狀態并且這個狀態又是流程的一個自然的輸出，那么這種方式就顯得尤為重要。
再組合使用上用于解決數據集成的數據庫輸出日志，日志和表的二象性的威力就更加明顯了。 從數據庫中抽取出來的變更日志可以按不同的形式索引到各種流處理器中，以關聯到事件流上。
在`Samza`和這些大量[實際例子](http://samza.apache.org/learn/documentation/0.7.0/container/state-management.html)中， 我們說明了這種風格的有狀態流處理管理的更多細節。
## [](https://github.com/oldratlee/translations/blob/master/log-what-every-software-engineer-should-know-about-real-time-datas-unifying/part3-logs-and-real-time-stream-processing.md#日志合并log-compaction)日志合并（`log compaction`）
[](https://github.com/oldratlee/translations/blob/master/log-what-every-software-engineer-should-know-about-real-time-datas-unifying/images/log_compaction_0.png)
當然，我們不能奢望一直保存著全部變更的完整日志。 除非想要使用無限空間，日志總是要清理。 為了讓討論更具體些，我會介紹一些`Kafka`這方面的實現。 在`Kafka`中，清理有兩種方式，取決于數據包括的是鍵值存儲的更新還是事件數據。 對于事件數據，`Kafka`支持僅維護一個窗口的數據。通常，窗口配置成幾天，但窗口也可以按空間大小來定。 對于鍵值存儲的更新，盡管完整日志的一個優點是可以回放以重建源系統的狀態（一般是另一個系統中重建）。
但是，隨著時間的推移，保持完整的日志會使用越來越多的空間，并且回放的耗時也會越來越長。 因此在`Kafka`中，我們支持不同類型的保留方式。 我們刪除過時的記錄（如這些記錄的主鍵最近更新過）而不是簡單的丟棄舊日志。 這樣做我們仍然保證日志包含了源系統的完整備份，但是現在我們不再重現原系統曾經的所有狀態，僅是最近的哪些狀態。 這一功能我們稱之為[日志合并](https://cwiki.apache.org/confluence/display/KAFKA/Log+Compaction)。
