編者注：本内容來自jay kreps所著的《我喜愛日志：事件資料、流計算處理和資料內建》一書的第三章。jay kreps是confluent的聯合創始人和ceo。在此之前，jay是領英的主要架構師之一，專注于資料基礎架構和資料驅動的産品。他是多個可擴充的資料系統空間的開源項目的作者之一，包括voldemort、azkaban、kafka和samza。
以下是原文：
到目前為止，我還僅僅隻是描述了一些把資料從一個地方拷貝到其他地方的多種的方法。然而，在存儲系統間挪動位元組并不是故事的結尾。實際上我們發現，“日志”是“流”的另外一種說法，而日志（的處理）是流計算處理的核心。
但是先等一下，到底什麼是流計算處理？
如果你是上世紀九十年代末和二十一世紀初的資料庫或者資料基礎設施産品的粉絲，你可能會把流計算處理和那些通過sql引擎或者用“流程圖”界面來進行資料驅動的處理過程聯系起來。
而如果你是追随着爆炸性增長的開源資料系統的人，你可能就會把流計算處理和諸如storm、akka、s4和samza這樣的系統聯系起來。很多人會把這些系統看成是一個異步消息處理系統，和那些基于叢集的rpc層上的應用沒什麼差別（事實上有些系統确實是這樣）。我還曾經聽有人過把流計算描述成一種模式，即立刻處理資料，随後就丢棄。
上述兩種觀點都有失偏頗。流計算處理與sql毫無關系；同時也不局限于實時處理系統。沒有任何的理由來限制你去用多種語言來處理昨天或者一個月以前的資料流；也沒有說你必須（或者應該）把獲得的原始資料丢棄掉。
我對流計算處理的看法則更加寬泛，即能做持續資料處理的基礎設施。我認為流計算處理的計算模型可以是如同mapreduce那樣的分布式處理架構一樣的通用，隻要它能提供低延遲的結果就可以。
而真正來驅動（或決定）處理模型的則是資料收集的方法。通過批次收集的資料則自然由按批次處理。對于持續流入的資料，就用持續的實時處理方式。
美國國家統計局的人口普查資料是一個按批次收集資料的好例子。統計局會定期啟動人口普查，派專人上門去挨家挨戶的收集美國公民的人口資料。這種方式對于在第一次普查開始的1790年（參見下圖）來說是有道理的。那時候的資料收集在本質上就是批次的，因為要騎馬去走訪，再在紙上填寫好統計記錄，再把這些記錄一批批地送到中心點去由人工來累加計算。而今時今日，在你給别人說這個人口普查的過程的時候，人們會立刻質疑為什麼我們不記錄一個出生和死亡的記錄，然後可以随時随地的用任何的粒度來計算人口總數。
第一次美國人口普查是批次收集資料的，因為受限于當時的技術條件。然而，在一個數字化、網絡化的世界裡，批次資料收集早已不是必須的了
當然，這（人口普查）是一個極端的例子。但是現在很多的資料傳輸過程依然依賴于定期的收集和批次化傳輸與內建。顯然批次收集資料的最自然的處理就是批次處理。随着這些過程逐漸被實時輸入收集所替代，我們也要相應的開始進行實時的資料處理，進而能平滑所需的處理資源，并降低延遲。
一個現在的網際網路公司并不需要有任何批次資料收集。網站産生的資料或是使用者行為資料或是資料庫的改變，而兩者都是持續發生的。事實上，如果你仔細想想，幾乎任何的業務的本質的機制都幾乎是一個持續的過程。如傑克 鮑爾所說，事件總是随時發生。出現是批次收集資料的情況，一般都是因為一些手工的過程或者是缺乏數字化，或者是因為曆史原因造成的沒法自動化或者數字化的材料。這時傳遞和對資料做出回報一般都非常慢，如果整個過程需要運送紙張并由人來做處理。剛剛實作自動化後，一般還是會保留原來的處理流程，是以即便是媒介發生了改變，而(這種)流程還是會持續很長時間。
每天運作的批次處理資料的産品經常是有效地模拟了用一天為時間視窗的持續計算。而底層的資料當然是總在改變。
上面的讨論有助于厘清對于流計算處理的常見誤解。通常認為某些種類的資料處理不适合用流計算系統來實作，而必須用批處理系統。我聽過的一個典型的例子就是計算百分位、最大值、均值和其他類似的需要用所有的資料來做的聚合統計。但是這往往帶來了某種誤解。确實，類似計算最大值這樣的塊操作需要用時間視窗内的所有資料。然後這樣的計算絕對能夠通過流計算系統來實作。事實上，如果你檢視早期的流計算的學術文章，一般它們完成的第一件事就是給出簡潔的語義來定義時間視窗，以便于針對于視窗的操作成為可能。
看到這裡，很容易能統一我對于流計算處理的觀點，即流計算是更寬泛。 它和是不是塊和非塊并沒有關系，僅僅隻是一個底層資料裡包含了時間定義的處理機制，并不要求對于處理的資料需要有一個靜态的快照。這意味着流計算處理系統以一個使用者控制的頻率來産出結果，而不是一直等到資料全部到達。從這個角度看，流計算是批次計算的一個更泛化的操作。考慮到現在實時資料的的普及，這應該是一個更重要的泛化。
為什麼這種傳統的對于流計算處理的觀點成為一個先進的應用。我認為最大的原因是因為缺乏實時資料收集的方法，進而讓持續處理成為某種理論上的想法。
我确實認為缺乏實時資料收集的方法是商用流計算處理系統的夢魇。它們的客戶依然是在做面向檔案的日複一日的etl和資料內建。建構流計算處理系統的公司一般專注于提供計算引擎來處理實時資料，但卻發現現實中很少有客戶有實時資料流。事實上在我在領英的早期時光，有個公司試圖賣給我們一套非常酷的流計算處理系統，但因為當時我們所有的資料都是按小時收集的檔案，是以我們所能想到的就是把這些小時檔案在每小時結束的時候喂給這個系統。這個公司的工程師發現這是一個非常常見的問題。唯一真實的例外就是金融界。在這個領域裡流計算處理有一些成功的案例，而恰恰是因為這個領域裡實時流資料才是主流，而如何處理這些實時資料流才是主要關注點。
即使是在健康的批處理生态系統裡，實際上流計算處理作為基礎架構類型的适用性也是很強的。它涵蓋了實時處理/相應業務和離線批處理業務的基礎架構上的鴻溝。對于現代的網際網路企業，我認為大約25%的代碼是關于這種鴻溝的。
現在發現日志（log）解決了流計算處理裡的一些非常關鍵的技術問題。後面我會陸續介紹這些問題，但其中最大的問題它解決的就是它讓資料成為了實時的多訂閱者的資料導入機制。
對那些希望能更多了解日志和流計算處理間的關系人，我們提供了開源的samza，一個專門為這些想法建構的實時流計算處理系統。在這個連結裡面我們很詳細地介紹了這些想法的應用。但這不是專門為了某個特定的流計算處理系統的，因為幾乎所有的主要流計算處理系統都和kafka有某種程度的內建，讓kafak來作為資料的日志來進行處理。
資料流圖
關于流計算處理的最有趣的方面就是它和一個流計算處理系統的内部機制沒有任何關系，想反的是，相關的是他擴充了我們前面資料內建讨論裡的資料源的觀點。我們主要讨論了主要資料源和主要資料的日志化。即事件和資料都是直接由各種應用運作中生成的。流計算處理讓我們可以也包含從其他資料源裡計算出的資料源。 這些計算出來的資料源對消費者而言與用來計算其的其他資料源沒什麼差別（請參看下圖）。
來自多日志的多路流處理圖
這些計算出來資料源可能會包含相當複雜和智力的成分在其處理過程裡，是以也是極具價值的。例如，谷歌在這裡描述了它是如何在一個流計算處理系統上重構它的網頁爬取、處理和建索引的管道的過程。這可能是這個行星上最複雜、最大規模的資料處理系統之一了。
是以什麼是流計算處理過程？對于我們的目的而言，一個流計算處理工作就是那些從日志中讀取并輸出到日志或其他系統的任務。那些作為輸入和輸出的日志把整個流程連接配接成了一個處理階段的圖。使用這種中心化日志的形式，你就能觀察所有機構的資料的擷取、轉換和流動，其實就是一系列日志以及從他們中讀出和寫入他們的過程。
一個流計算處理過程并不必需要有一個時髦的架構。它可以是任何一個或多個讀取和寫入日志的過程。額外的基礎架構和支援能夠幫助管理和擴充這種近乎實時的處理過程程式，而這也就是流計算架構所做的。
為什麼你在所有的流計算處理裡需要日志？為什麼不是讓處理單元通過簡單的tcp或者其他輕量級的消息協定來更直接的通信？有多個理由來支援這一（日志）模式。
首先，這種模式可以讓每個資料集都能為多訂閱者所用。每個流處理過程的輸入對任何需要的處理器都可用；同時每個輸出也都對任何需要的都可用。這一點不僅對生産資料流很好用，而且也在複雜的資料處理管道裡調試和監控階段很有幫助。能快速的進入一個輸出流并檢查它的有效性，同時計算一些監控的統計資料，或者僅僅隻是看看資料長什麼樣，這些都使得開發變的非常有可追蹤性。
其次，這樣使用日志能確定每個資料消費者處理過程中順序可以被保留。某些事件資料可能被按時間戳松散地排序了，但是不是每種事件資料都這樣。考慮從來自資料庫的一個更新流，我們可能有一系列的流處理任務來處理這些資料并準備為搜尋索引來做索引。如果對同一個記錄同時做兩次更新，那麼我們最後可能在索引的最終結果出錯。
這樣使用日志的最後一個可能也是最重要（可探讨）的原因是它提供了緩存和對每個處理過程的隔離。如果一個處理器産生結果的速度比它後續的消費程式的處理能力快，我們可以有三種選擇：
我們可以先暫停上遊的處理任務，直到下遊的任務可以處理。如果隻用tcp而沒有使用日志，這種情況是最可能發生的。
我們就把資料丢棄掉。
我們可以在兩個處理任務間緩存資料。
丢棄資料在某些場合可能沒什麼。但是基本都是不可接受的，也從來不被希望這樣做。
暫停（上遊）處理聽起來似乎是一個可接受的選擇。但實際中這會成為一個很大的問題。考慮到我們需要的不僅僅是對單一的應用流程建立模型，而是為整個機建構立全套的資料流模型。這就将不可避免的形成一個複雜的資料處理流網絡，由不同的部門團隊的不同的資料處理器來組成，并支援不同的sla。在這樣複雜的資料處理網絡裡，如果因為後續處理能力不足或者失敗而導緻上遊的資料産生器被暫停，這都會級聯地影響上遊資料流程式，進而使得的很多處理器都被暫停。
這樣看來唯一可用的選擇：緩存。日志可以是非常非常大的緩存，可以讓處理程式被重新開機，或者即使失效了也不會影響處理圖裡的其他部分。這也意味着某個資料消費程式可以停機很長時間，而不會影響上遊的程式。隻要在它重新開機後能及時處理完緩存的資料，大家都皆大歡喜。
在其他地方，這也不是一個不尋常的模式。巨大、複雜的mapreuce流就使用了檔案作為檢查點，并共享他們的中間處理結果。巨大、複雜的sql處理管道也是建立了很多中間的臨時表。這裡僅僅隻是運用了這種模式的抽象-日志-使得它适合于處理運動中的資料。
storm和samza是兩種基于這個模式建構的流技術處理系統，也能使用kafka或者其他類似的系統作為他們的日志部分。
處理資料：lambda架構和一個可替換的方案
一個基于這種日志資料模型的有趣的應用就是lambda架構，由内森·馬茲提出。他寫了一個廣為傳播的部落格（《如何打敗cap定理》），其中介紹把流處理系統和批次處理相結合的方法。這個架構被證明是一個非常流行的想法。已經有專門的網站和書籍了。
什麼是lambda架構？如何運用？
lambda架構一般類似于下圖。
它工作的方式是不可變的一系列資料記錄被采集，并同時并行地送給批處理和流處理系統。資料轉換的邏輯被實作兩次，一次是在批處理系統裡，一次是在流處理系統裡。然後把二者處理的結果在查詢的時候合并，給出一個完整的答案。
這個方式有很多中變形，這裡我是有意地簡化了許多。例如你可以使用多種類似的系統，如kafka、storm和hadoop。而大家也經常會使用兩種不同的資料庫來存儲輸出結果，一種是專門為實時處理優化的資料庫，而其他的則是為批處理所準備的。
lambda架構的優點：
lambda架構強調保留原始輸入資料不變。我認為這是一個非常重要的特性。處理複雜資料流的能力得到了很大的加強，因為能夠看到資料是什麼樣和輸出是什麼樣。
我還喜歡這個架構強調了重複處理資料的問題。能重複處理資料對流計算系統來說是一個重大挑戰，但卻經常被忽略。
對于重複處理，我的意思是再次處理輸入的資料進而再次計算出結果。這是一個太明顯，但也經常被忽略的需求。代碼總是在變。是以如果你的代碼已經從輸入流中計算出了結果，當代碼改變後，你需要再次計算輸出來檢查代碼修改的效果。
為什麼代碼會變？也許是因為你的應用在演進，你又想計算一些之前不需要的新的輸出項。或者是你發現了一個代碼缺陷并修好了。無論如何，當這件事發生的時候，你就需要重新計算你的輸出結果。我發現很多的人試圖去建構一個實時資料處理系統，但根本不去仔細思考這個問題，并最終導緻系統不能很快的演進，僅是因為沒有一個好的方法來解決重複處理的需求。
lambda架構的一個問題就是需要維護兩套複雜的分布式系統的代碼，這看起來就很頭疼。但我不認為這個問題不能解決。
如storm和hadoop這樣的分布式架構的程式設計是很複雜。不可避免的是代碼最後會專門為所使用的架構而特殊地建構。由此導緻的運維的複雜性是被所有使用這個架構的人所一緻同意的。
解決這一問題的一個方法是對實時和批次架構抽象出一種語言或架構。用這個進階架構來寫你的代碼，然後去“編譯”成或是流計算處理代碼，或是mapreduce的批處理代碼。summingbird就是這樣做的一個架構。它确實讓事情好了一些，但我不認為它真正解決了問題。最終，即便你能避免為應用寫兩套代碼，運維兩套系統的負擔還是很重的。新的抽象僅僅是提供了兩套系統的交集所支援的特征。更糟的是，使用這一統一的架構就把整個生态系統裡那些使得hadoop非常強大的工具和語言（如hive、pig、crunch、cascading、oozie等）給排除在外了。
打個比方，想想那個使跨資料庫對象關系映射（orm）透明化的臭名昭著的難題。而這也還是對非常相同的系統進行抽象來用标準接口語言提供相同的能力哦。那麼去抽象化兩個完全不同的建構于剛剛穩定化的分布式系統上的程式設計模式将會更加的難。
一個備選方案
作為一個基礎架構的設計者，我認為真正有意義的問題是：為什麼不去改進流計算系統來解決所有的問題集？為什麼你需要再粘貼一個别的系統（批處理）？為什麼你不去把實時處理和在代碼改變時需要的重複處理一起解決？流計算處理系統已經有了并行的概念，為什麼不是去通過增加并行性來解決重複處理的問題，并很快的再現曆史？答案是你可以這麼做。我認為這就是如果你需要建構這樣一個系統的一個合理的備選方案。
當我和别人讨論這個思想的時候，有時他們會告訴我流計算處理對于高吞吐率處理曆史資料并不合适。但這是他們基于已經使用的系統的缺陷的直覺反應。這些系統要不就是很難擴充，或者是根本就沒存曆史資料。但沒有理由認為這就是對的。流計算處理系統的基本抽象就是資料流的有向無環圖（dag）。這和傳統的資料倉庫（如volcano）的基本抽象是一樣的，并和mapreduce的後繼tez的基本底層抽象也是一樣。流計算處理僅僅是對這一資料流模型的泛化，即對中間結果提供檢查點（checkpointing）并持續地輸出到最終的使用者。
那麼我們怎麼才能從我們的流處理任務裡完成重複處理？我最喜歡的方法其實非常的簡單。
1. 使用kafka或者其他的一些能幫你儲存你想重複處理的資料的全部日志的系統，這些系統還要能支援多訂閱者功能。例如，如果你想重複處理之前30天的資料，那就把kafka的儲存時間設成30天。
2. 當你想重複處理資料時，啟動第二個你的流計算系統的執行個體，從保留資料的開始再次處理這些資料，不過把結束輸出到新的表。
3. 當第二個執行個體已經可以趕上現有資料的進度了，就把應用定向到從新的輸出表裡讀取資料。
4. 停止原有版本的任務執行個體，并删除舊的輸出表。
這個架構類似下圖所示。
lambda架構的一個備選替換方案，移除了批處理系統
與lambda架構不同，這個方法裡你隻是當你代碼改變而确實需要再計算結果的時候你才需要重複處理。當然這樣的重複計算隻是用你代碼的改進版本，使用相同的架構，并處理相同的資料。
很自然的，你希望能多給你的并行的重複處理任務更多的資源以便于讓它能非常快地完成。
當然，你可以進一步優化這個方法。很多情況下你可以把兩個輸出表合并。然而我認為讓兩個輸出表并存一段時間是有不少好處的。這可以及時回退回舊的邏輯，而你所需要做的僅僅隻是把應用再重定向到舊的表。另外，對于一些非常重要的場景裡，你可以使用自動的a/b測試或者多臂強盜算法來控制這個切換，保證新的代碼确實是比舊有邏輯有改進而不是變的更差了。
需要注意的是，這個方法并不是說你的資料不能輸出到hadoop裡。它僅僅是說你不必要在hadoop裡面做重複處理。kafka和hadoop有非常好的內建，是以從中導出任何kafka的topic都很簡單。通常把一個流計算處理任務的輸出甚至是中間結果鏡像到hadoop中，進而讓一些例如hive這樣的分析工具來處理，或者是作為其他人的輸入，或是為離線資料處理流服務。這都是很有用的。
我們已經記錄了如何實作這個方法，包括使用samza實作重複處理的架構的其他變形。
這兩種方法所對應的效率和資源之間的權衡是值得讨論的。lambda架構需要同時一直運作重複處理和實時處理任務。而我提出的方法僅僅隻是在需要重複處理的時候才運作第二個任務的執行個體。然而我的提議要求有臨時的額外的輸出資料庫的存儲。同時資料庫也要能支援這樣大容量的寫操作。兩種情況下，重複處理所造成的額外任務都可以被平均分布出去。如果你有很多這樣的任務，他們不需要一下都重複處理完。是以在一個共享的叢集裡，如果有很多個這樣的任務，你需要預留一部分容量來處理這些随時會發生的任務。
我提議方法的真正的優勢不是效率，而是能讓大家在一個單一的處理架構裡開發、測試、調試和運維他們的系統。
是以在簡介是很重要的場景裡，可以把我的方法作為lambda架構的一個備選方案。
◆ ◆ ◆
有狀态的實時處理
日志和流計算處理之間的關系不僅僅限于重複處理。如果實際的流計算處理系統需要維護狀态資訊，這時使用日志就可以有另一個新的用處了。
一些實時流處理系統僅僅是無狀态的一次性資料轉換。但是很多場景下都是比較複雜的計數、彙聚或視窗間的連接配接等操作。例如，你可能希望對事件流（比如點選流）進行增強，比如通過連接配接點選流和使用者帳号資料庫來給點選加上使用者資訊。不可避免的，類似這樣的處理最終都會需要儲存某種程度的狀态資訊。例如當計數的時候，你需要保留此前的計數值。這樣的狀态資訊怎麼樣才能保留下來當處理器自身會發生失效？
最簡單的方法就是把狀态資訊放到記憶體裡。然而如果處理器崩潰了，這個資訊就丢失了。如果狀态資訊僅僅是在一個視窗裡維護的，這個處理器就可以從這個視窗開始的點重新再來。但是如果計算一小時的計數，這個方法可能就不行了。
另外一個方法就是把所有的狀态資訊都存儲到一個異地系統，并通過網絡擷取。這個方法的問題是沒有本地化的資料，還需要很多的網絡傳輸。
我們怎麼能支援一些如把一張表分片到處理裡的操作？
回顧之前對于表和日志的二進制性的讨論，這給我們提供了能把流轉換成表并和我們的處理并存的方法。這還提供了一個對于表失效的解決機制。
流處理器可以把它的狀态保持到一個本地的表或者索引裡，如dbd和rocksdn，或者一些更不尋常的機制，比如lucene或fastbit索引等。這些存儲的内容是從輸入流裡導入的（可能是首先做一個強制轉換後）。它對本地索引可以記錄下修改日志（changelog），并儲存這些修改日志，進而在系統奔潰或重新開機後可以恢複出狀态資訊來。這就可以提供一個通用的機制來儲存狀态資訊在一個索引類型的本地内。
當（流）處理過程失效，它就從修改日志裡恢複索引。日志在這裡就變成了把本地的狀态資訊轉換成一個增長的随時記錄的備份。
這個狀态管理的方法有一個優雅的特性，即處理器的狀态也被維護成了一個日志。我們可以把這個日志想成是資料庫表的修改日志。事實上，處理器有一些伴随着它的非常類似于同分片表的東西。因為狀态本身就是日志，其他的處理器也可以訂閱它，這就可以非常有用。比如整個處理過程的目标就是更新輸出的最終狀态的場景。
當把從資料庫裡輸出的日志結合起來看，日志/表二進制性的威力就非常清晰了。修改日志（changlog）可以從資料庫裡抽取，并被不同的流處理器用不同的方式檢索來和事件流所連接配接。
我們提供了在samza裡面使用這種類型的狀态管理的細節，以及很多實際的例子。
當然，我們不能希望能保持所有時間的狀态改變的完整日志。除非你有無限的空間，不然日志總是要被清理的。我會介紹如何在kafka裡面實作這個功能的。
在kafka裡，清理有兩個選擇，取決于資料是僅包含純事件資料還是鍵值化的更新。對于事件資料，我的意思是沒有相關的情形發生，比如網頁浏覽、點選或者其他你會在一個應用的日志裡發現的東西。對于鍵值化的更新，我的意思是事件有特别記錄的狀态改變，而這被用某些鍵值所識别。資料庫的修改就是一個典型的鍵值化更新的例子。
對于事件資料，kafka支援儲存資料的視窗。這個視窗可以是用時間（以天）或者是空間（以gb）為機關。大部分人僅僅隻是使用它預設的一個星期為儲存視窗。如果你希望有無限的儲存期，就把這個視窗設成無限，你的資料就永遠不會丢失。
然而，對于鍵值化的資料，一個完整日志的非常好的特征就是你可以重制源系統的狀态。即，如果你有這個修改的日志，你可以在另外一個資料庫裡再現這個表，并重建這個表的任意時間點的狀态。這對于不同系統也适用。你可以在另外一個資料庫裡再現源資料庫裡的更新，并維護資料的主鍵（一個搜尋的索引、一個本地庫等等）。
然而，随着時間的延長，儲存完整的日志會消耗越來越多的空間，再現過程也會越來越久。是以在kafka裡，為了支援這種應用場景，我們支援不同類型的儲存。其中一個例子就展示在下圖裡。不是簡單地完全丢棄舊的日志，我們收集了日志末尾的部分裡過時的記錄作為垃圾。任何在日志末尾的記錄有最近的更新就會适合于清理（隻保留最新的更新）。這麼做就可以保證日志儲存了一個源系統的完整的備份，但是我們現在不必在完全重建所有的之前的狀态了，而僅僅隻是最近更新的狀态。我們把這個特性稱為日志壓縮。
日志壓縮可以確定日志裡隻保留每個鍵值的最新的更新。這對模型更新成可變資料的日志是很有用的
原文釋出時間為：2016-06-20
本文來自雲栖社群合作夥伴“大資料文摘”，了解相關資訊可以關注“bigdatadigest”微信公衆号