LinkedIn大數(shù)據(jù)后臺(tái)是如何運(yùn)作的_數(shù)據(jù)分析師

第一部分：Log是什么?

　　第二部分：數(shù)據(jù)集成

　　第三部分：日志和實(shí)時(shí)流處理

　　第四部分：系統(tǒng)建設(shè)

　　第三部分：日志和實(shí)時(shí)流處理

　　到此為止，我只是描述從端到端數(shù)據(jù)復(fù)制的理想機(jī)制。但是在存儲(chǔ)系統(tǒng)中搬運(yùn)字節(jié)不是所要講述內(nèi)容的全部。最終我們發(fā)現(xiàn)日志是流的另一種說(shuō)法，日志是流處理的核心。

　　但是，等等，什么是流處理呢?

　　如果你是90年代晚期或者21世紀(jì)初數(shù)據(jù)庫(kù)文化或者數(shù)據(jù)基礎(chǔ)架構(gòu)產(chǎn)品的愛(ài)好者，那么你就可能會(huì)把流處理與建創(chuàng)SQL引擎或者創(chuàng)建“箱子和箭頭”接口用于事件驅(qū)動(dòng)的處理等聯(lián)系起來(lái)。

　　如果你關(guān)注開(kāi)源數(shù)據(jù)庫(kù)系統(tǒng)的大量出現(xiàn)，你就可能把流處理和一些開(kāi)源數(shù)據(jù)庫(kù)系統(tǒng)關(guān)聯(lián)起來(lái)，這些系統(tǒng)包括了：Storm,Akka,S4和Samza.但是大部分人會(huì)把這些系統(tǒng)作為異步消息處理系統(tǒng)，這些系統(tǒng)與支持群集的遠(yuǎn)程過(guò)程調(diào)用層的應(yīng)用沒(méi)什么差別(而事實(shí)上在開(kāi)源數(shù)據(jù)庫(kù)系統(tǒng)領(lǐng)域某些方面確實(shí)如此)。

　　這些視圖都有一些局限性。流處理與SQL是無(wú)關(guān)的。它也局限于實(shí)時(shí)流處理。不存在內(nèi)在的原因限制你不能處理昨天的或者一個(gè)月之前的流數(shù)據(jù)，且使用多種不同的語(yǔ)言表達(dá)計(jì)算。

　　我把流處理視為更廣泛的概念：持續(xù)數(shù)據(jù)流處理的基礎(chǔ)架構(gòu)。我認(rèn)為計(jì)算模型可以像MapReduce或者分布式處理架構(gòu)一樣普遍，但是有能力處理低時(shí)延的結(jié)果。

　　處理模型的實(shí)時(shí)驅(qū)動(dòng)是數(shù)據(jù)收集方法。成批收集的數(shù)據(jù)是分批處理的。數(shù)據(jù)是不斷收集的，它也是按順序不斷處理的。

　　美國(guó)的統(tǒng)計(jì)調(diào)查就是成批收集數(shù)據(jù)的良好典范。統(tǒng)計(jì)調(diào)查周期性的開(kāi)展，通過(guò)挨門挨戶的走訪，使用蠻力發(fā)現(xiàn)和統(tǒng)計(jì)美國(guó)的公民信息。1790年統(tǒng)計(jì)調(diào)查剛剛開(kāi)始時(shí)這種方式是奏效的。那時(shí)的數(shù)據(jù)收集是批處理的，它包括了騎著馬悠閑的行進(jìn)，把信息寫在紙上，然后把成批的記錄傳送到人們統(tǒng)計(jì)數(shù)據(jù)的中心站點(diǎn)?，F(xiàn)在，在描述這個(gè)統(tǒng)計(jì)過(guò)程時(shí)，人們立即會(huì)想到為什么我們不保留出生和死亡的記錄，這樣就可以產(chǎn)生人口統(tǒng)計(jì)信息這些信息或是持續(xù)的或者是其它維度的。

　　這是一個(gè)極端的例子，但是大量的數(shù)據(jù)傳送處理仍然依賴于周期性的轉(zhuǎn)儲(chǔ)，批量轉(zhuǎn)化和集成。處理大容量轉(zhuǎn)儲(chǔ)的唯一方法就是批量的處理。但是隨著這些批處理被持續(xù)的供給所取代，人們自然而然的開(kāi)始不間斷的處理以平滑的處理所需資源并且消除延遲。

　　例如LinkedIn幾乎沒(méi)有批量數(shù)據(jù)收集。大部分的數(shù)據(jù)或者是活動(dòng)數(shù)據(jù)或者是數(shù)據(jù)庫(kù)變更，這兩者都是不間斷發(fā)生的。事實(shí)上，你可以想到的任何商業(yè)，正如：Jack Bauer告訴我們的，低層的機(jī)制都是實(shí)時(shí)發(fā)生的不間斷的流程事件。數(shù)據(jù)是成批收集的，它總是會(huì)依賴于一些人為的步驟，或者缺少數(shù)字化或者是一些自動(dòng)化的非數(shù)字化流程處理的遺留信息。當(dāng)傳送和處理這些數(shù)據(jù)的機(jī)制是郵件或者人工的處理時(shí)，這一過(guò)程是非常緩慢的。首輪自動(dòng)化總是保持著最初的處理形式，它常常會(huì)持續(xù)相當(dāng)長(zhǎng)的時(shí)間。

　　每天運(yùn)行的批量處理作業(yè)常常是模擬了一種一天的窗口大小的不間斷計(jì)算。當(dāng)然，低層的數(shù)據(jù)也經(jīng)常變化。在LinkedIn,這些是司空見(jiàn)貫的，并且使得它們?cè)?a href='/map/hadoop/' style='color:#000;font-size:inherit;'>Hadoop運(yùn)轉(zhuǎn)的機(jī)制是有技巧的，所以我們實(shí)施了一整套管理增量的Hadoop工作流的架構(gòu)。

　　由此看來(lái)，對(duì)于流處理可以有不同的觀點(diǎn)。流處理包括了在底層數(shù)據(jù)處理的時(shí)間概念，它不需要數(shù)據(jù)的靜態(tài)快照，它可以產(chǎn)生用戶可控頻率的輸出，而不用等待數(shù)據(jù)集的全部到達(dá)。從這個(gè)角度上講，流處理就是廣義上的批處理，隨著實(shí)時(shí)數(shù)據(jù)的流行，會(huì)兒更加普遍。

　　這就是為什么從傳統(tǒng)的視角看來(lái)流處理是利基應(yīng)用。我個(gè)人認(rèn)為最大的原因是缺少實(shí)時(shí)數(shù)據(jù)收集使得不間斷的處理成為了學(xué)術(shù)性的概念。

　　我想缺少實(shí)時(shí)數(shù)據(jù)收集就像是商用流處理系統(tǒng)注定的命運(yùn)。他們的客戶仍然需要處理面向文件的、每日批量處理ETL和數(shù)據(jù)集成。公司建設(shè)流處理系統(tǒng)關(guān)注的是提供附著在實(shí)時(shí)數(shù)據(jù)流的處理引擎，但是最終當(dāng)時(shí)極少數(shù)人真正使用了實(shí)時(shí)數(shù)據(jù)流。事實(shí)上，在我在LinkedIn工作的初期，有一家公司試圖把一個(gè)非常棒的流處理系統(tǒng)銷售給我們，但是因?yàn)楫?dāng)時(shí)我們的全部數(shù)據(jù)都按小時(shí)收集在的文件里，當(dāng)時(shí)我們提出的最好的應(yīng)用就是在每小時(shí)的最后把這些文件輸入到流處理系統(tǒng)中。他們注意到這是一個(gè)普遍性的問(wèn)題。這些異常證明了如下規(guī)則：流處理系統(tǒng)要滿足的重要商業(yè)目標(biāo)之一是：財(cái)務(wù)， 它是實(shí)時(shí)數(shù)據(jù)流已具備的基準(zhǔn)，并且流處理已經(jīng)成為了瓶頸。

　　甚至于在一個(gè)健康的批處理系統(tǒng)中，流處理作為一種基礎(chǔ)架構(gòu)的實(shí)際應(yīng)用能力是相當(dāng)廣泛的。它跨越了實(shí)時(shí)數(shù)據(jù)請(qǐng)求-應(yīng)答服務(wù)和離線批量處理之間的鴻溝。現(xiàn)在的互聯(lián)網(wǎng)公司，大約25%的代碼可以劃分到這個(gè)類型中。

　　最終這些日志解決了流處理中絕大部分關(guān)鍵的技術(shù)問(wèn)題。在我看來(lái)，它所解決的最大的問(wèn)題是它使得多訂閱者可以獲得實(shí)時(shí)數(shù)據(jù)。對(duì)這些技術(shù)細(xì)節(jié)感興趣的朋友，我們可以用開(kāi)源的Samza,它是基于這些理念建設(shè)的一個(gè)流處理系統(tǒng)。這些應(yīng)用的更多技術(shù)細(xì)節(jié)我們?cè)诖宋臋n中有詳細(xì)的描述。

　　數(shù)據(jù)流圖

　　流處理最有趣的角度是它與流處理系統(tǒng)內(nèi)部無(wú)關(guān)，但是與之密切相關(guān)的是如何擴(kuò)展了我們談到的早期數(shù)據(jù)集成的數(shù)據(jù)獲取的理念。我們主要討論了基礎(chǔ)數(shù)據(jù)的獲取或日志–事件和各類系統(tǒng)執(zhí)行中產(chǎn)生的數(shù)據(jù)等。但是流處理允許我們包括了計(jì)算其它數(shù)據(jù)的數(shù)據(jù)。這些衍生的數(shù)據(jù)在消費(fèi)者看來(lái)與他們計(jì)算的原始數(shù)據(jù)沒(méi)什么差別。這些衍生的數(shù)據(jù)可以按任意的復(fù)雜度進(jìn)行壓縮。

　　讓我們?cè)偕钊胍徊?。我們的目?biāo)是：流處理作業(yè)可以讀取任意的日志并把日志寫入到日志或者其它的系統(tǒng)中。他們用于輸入輸出的日志把這些處理關(guān)聯(lián)到一組處理過(guò)程中。事實(shí)上，使用這種樣式的集中日志，你可以把組織全部的數(shù)據(jù)抓取、轉(zhuǎn)化和工作流看成是一系列的日志和寫入它們的處理過(guò)程。

　　流處理器根本不需要理想的框架：它可能是讀寫日志的任何處理器或者處理器集合，但是額外的基礎(chǔ)設(shè)施和輔助可以提供幫助管理處理代碼。

　　日志集成的目標(biāo)是雙重的：

　　首先，它確保每個(gè)數(shù)據(jù)集都有多個(gè)訂閱者和有序的。讓我們回顧一下?tīng)顟B(tài)復(fù)制原則來(lái)記住順序的重要性。為了使這個(gè)更加具體，設(shè)想一下從數(shù)據(jù)庫(kù)中更新數(shù)據(jù)流–如果在處理過(guò)程中我們把對(duì)同一記錄的兩次更新重新排序，可能會(huì)產(chǎn)生錯(cuò)誤的輸出。 TCP之類的鏈接僅僅局限于單一的點(diǎn)對(duì)點(diǎn)鏈接，這一順序的持久性要優(yōu)于TCP之類的鏈接，它可以在流程處理失敗和重連時(shí)仍然存在。

　　第二，日志提供了流程的緩沖。這是非?；A(chǔ)的。如果處理流程是非同步的，那么上行生成流數(shù)據(jù)的作業(yè)比下行消費(fèi)流數(shù)據(jù)的作業(yè)運(yùn)行的更快。這將會(huì)導(dǎo)致處理流程阻塞，或者緩沖數(shù)據(jù)，或者丟棄數(shù)據(jù)。丟棄數(shù)據(jù)并不是可行的方法，阻塞將會(huì)導(dǎo)致整個(gè)流程圖立即停止。 日志實(shí)際上是一個(gè)非常大的緩沖，它允許流程重啟或者停止但不會(huì)影響流程圖其它部分的處理速度。如果要把數(shù)據(jù)流擴(kuò)展到更大規(guī)模的組織，如果處理作業(yè)是由多個(gè)不同的團(tuán)隊(duì)提供的，這種隔離性是極其重的。我們不能容忍一個(gè)錯(cuò)誤的作業(yè)引發(fā)后臺(tái)的壓力，這種壓力會(huì)使得整個(gè)處理流程停止。

　　Storm和Sama這兩者都是按非同步方式設(shè)計(jì)的，可以使用Kafka或者其它類似的系統(tǒng)作為它們的日志。

　　有狀態(tài)的實(shí)時(shí)流處理

　　一些實(shí)時(shí)流處理在轉(zhuǎn)化時(shí)是無(wú)狀態(tài)的記錄。在流處理中大部分的應(yīng)用會(huì)是相當(dāng)復(fù)雜的統(tǒng)計(jì)、聚合、不同窗口之間的關(guān)聯(lián)。例如有時(shí)人們想擴(kuò)大包含用戶操作信息的事件流(一系列的單擊動(dòng)作)–實(shí)際上關(guān)聯(lián)了用戶的單擊動(dòng)作流與用戶的賬戶信息數(shù)據(jù)庫(kù)。不變的是這類流程最終會(huì)需要由處理器維護(hù)的一些狀態(tài)信息。例如數(shù)據(jù)統(tǒng)計(jì)時(shí)，你需要統(tǒng)計(jì)到目前為止需要維護(hù)的計(jì)數(shù)器。如果處理器本身失敗了，如何正確的維護(hù)這些狀態(tài)信息呢?

　　最簡(jiǎn)單的替換方案是把這些狀態(tài)信息保存在內(nèi)存中。但是如果流程崩潰，它就會(huì)丟失中間狀態(tài)。如果狀態(tài)是按窗口維護(hù)的，流程就會(huì)回退到日志中窗口開(kāi)始的時(shí)間點(diǎn)上。但是，如果統(tǒng)計(jì)是按小時(shí)進(jìn)行的，那么這種方式就會(huì)變得不可行。

　　另一個(gè)替換方案是簡(jiǎn)單的存儲(chǔ)所有的狀態(tài)信息到遠(yuǎn)程的存儲(chǔ)系統(tǒng)，通過(guò)網(wǎng)絡(luò)與這些存儲(chǔ)關(guān)聯(lián)起來(lái)。這種機(jī)制的問(wèn)題是沒(méi)有本地?cái)?shù)據(jù)和大量的網(wǎng)絡(luò)間通信。

　　我們?nèi)绾沃С痔幚磉^(guò)程可以像表一樣分區(qū)的數(shù)據(jù)呢?

　　回顧一下關(guān)于表和日志二相性的討論。這一機(jī)制提供了工具把數(shù)據(jù)流轉(zhuǎn)化為與處理過(guò)程協(xié)同定位的表，同時(shí)也提供了這些表的容錯(cuò)處理的機(jī)制。

　　流處理器可以把它的狀態(tài)保存在本地的表或索引–bdb,或者leveldb,甚至于類似于Lucene 或fastbit一樣不常見(jiàn)的索引。這些內(nèi)容存儲(chǔ)在它的輸入流中(或許是使用任意的轉(zhuǎn)化)。生成的變更日志記錄了本地的索引，它允許存儲(chǔ)事件崩潰、重啟等的狀態(tài)信息。流處理提供了通用的機(jī)制用于在本地輸入流數(shù)據(jù)的隨機(jī)索引中保存共同分片的狀態(tài)。

　　當(dāng)流程運(yùn)行失敗時(shí)，它會(huì)從變更日志中恢復(fù)它的索引。每次備份時(shí)，日志把本地狀態(tài)轉(zhuǎn)化成一系列的增量記錄。

　　這種狀態(tài)管理的方法有一個(gè)優(yōu)勢(shì)是把處理器的狀態(tài)也做為日志進(jìn)行維護(hù)。我們可以把這些日志看成與數(shù)據(jù)庫(kù)表相對(duì)應(yīng)的變更日志。事實(shí)上，這些處理器同時(shí)維護(hù)著像共同分片表一樣的表。因?yàn)檫@些狀態(tài)它本身就是日志，其它的處理器可以訂閱它。如果流程處理的目標(biāo)是更新結(jié)點(diǎn)的最后狀態(tài)，這種狀態(tài)又是流程的輸出，那么這種方法就顯得尤為重要。

　　為了數(shù)據(jù)集成，與來(lái)自數(shù)據(jù)庫(kù)的日志關(guān)聯(lián)，日志和數(shù)據(jù)庫(kù)表的二象性就更加清晰了。變更日志可以從數(shù)據(jù)庫(kù)中抽取出來(lái)，日志可以由不同的流處理器(流處理器用于關(guān)聯(lián)不同的事件流)按不同的方式進(jìn)行索引。

　　我們可以列舉在Samza中有狀態(tài)流處理管理的更多細(xì)節(jié)和大量實(shí)用的例子。

　　日志壓縮

　　當(dāng)然，我們不能奢望保存全部變更的完整日志。除非想要使用無(wú)限空間，日志不可能完全清除。為了澄清它，我們?cè)賮?lái)聊聊Kafka的實(shí)現(xiàn)。在Kafka中,清理有兩種選擇，這取決于數(shù)據(jù)是否包括關(guān)鍵更新和事件數(shù)據(jù)。對(duì)于事件數(shù)據(jù)，Kafka支持僅維護(hù)一個(gè)窗口的數(shù)據(jù)。通常，配置需要一些時(shí)間，窗口可以按時(shí)間或空間定義。雖然對(duì)于關(guān)鍵數(shù)據(jù)而言，完整日志的重要特征是你可以重現(xiàn)源系統(tǒng)的狀態(tài)信息，或者在其它的系統(tǒng)重現(xiàn)。

　　隨著時(shí)間的推移，保持完整的日志會(huì)使用越來(lái)越多的空間，重現(xiàn)所耗費(fèi)的時(shí)間越來(lái)越長(zhǎng)。因些在Kafka中,我們支持不同類型的保留。我們移除了廢棄的記錄(這些記錄的主鍵最近更新過(guò))而不是簡(jiǎn)單的丟棄舊日志。我們?nèi)匀槐ＷC日志包含了源系統(tǒng)的完整備份，但是現(xiàn)在我們不再重現(xiàn)原系統(tǒng)的全部狀態(tài)，而是僅僅重現(xiàn)最近的狀態(tài)。我們把這一特征稱為日志壓縮。