數據收集工具的設計與最佳實踐

最新 10-14

策劃 & 編輯｜Natalie

筆者之前在《七牛大數據平台架構演進之路與使用實戰》中提到了已經開源的數據收集工具 logkit。本文將深入介紹數據收集的設計思路以及大數據收集背後的細節，為大家提供大數據實戰中第一步數據收集的最佳實踐。

本文是講解大數據平台組件系列文章的第一篇。眾所周知，大數據生態很複雜，大數據平台的每一個環節都涉及眾多開源組件，對個人來說，全部學會需要耗費很多精力和時間，InfoQ期望通過這一系列文章幫助大家縮短學習時間、更好地了解大數據平台各個模塊的設計思路及實踐經驗。

數據收集工具對比

目前社區已經不乏大量優秀的數據收集工具，如有名的 Elastic Stack（Elasticsearch、Logstash、Kibana）中的 Logstash；CNCF 基金會裡面有名的 Fluentd；InfluxData 公司 TICK Stack 中的 Telegraf；Google 出品為 Kubernetes 定製的 cAdvisor；Apache 基金會中的頂級項目 Flume。除了早期誕生的諸如 Fluentd、Flume 等項目，其他項目都是為特定的平台業務定製而成，然後在隨後的開源中不斷進化，變得更為通用。所以針對特定業務量身定製一款數據收集工具，是一個較為普遍的需求，也是出現如此多「輪子」的主要原因。

讓我們先來看看這幾種知名開源數據收集工具有哪些特點。

Flume: 毋庸置疑，在流式數據處理的場景中，Flume 絕對是開源產品中的不二選擇。其架構分為 Source、Channel、Sink 三部分，分別負責從上游服務端獲取數據、暫存數據以及解析並發送到下游。Flume 尤以靈活的擴展性和強大的容錯處理能力著稱，非常適合在大數據量的情況下做數據解析、中轉以及上下游適配的工作。

另一方面，Flume 也有一些缺陷，如解析與發送都耦合在 Sink 模塊，用戶在編寫 Sink 插件時不得不編寫解析的邏輯，無法復用一些常規的解析方式；依賴 JVM 運行環境，作為服務端程序可以接受，但是部署和運行一個數據收集客戶端程序則變得相對笨重；Flume 的配置融合了 Channel 部分，基本配置並不簡單，用戶想用起來需要的前置知識較多。

Logstash: 隨著 Elastic Stack 廣受熱捧，Logstash 自然也成為了技術圈家喻戶曉的工具，而 Logstash 本身的強大功能也確實名副其實，其架構分為 Inputs、Filters 以及 Outputs 三部分。Inputs 作為所有輸入端的集合，包含了各類數據輸入插件；Filters 包括解析與數據轉換兩部分的插件集合，其中就包含了大名鼎鼎的 Grok 解析方式，幾乎可以解析所有類型的數據；Outputs 則是輸出端的集合。毫無疑問，Logstash 幾乎是使用 Elastic Stack 方案時作為數據收集的唯一選擇。

但是同樣的，Logstash 也是基於 JVM 運行的，作為一個客戶端程序相對較重；當你希望把它作為一個 agent 收集一些機器的基本信息時，它也無能為力。於是除了 Logstash 之外 Elastic Stack 家族中又增加了 beats 這個成員，但是如果僅僅選擇 beats，其功能又太過單薄。

Fluentd: Fluentd 也是數據收集界的老牌重量級選手，如果你玩容器、玩 Kubernetes，那麼就一定聽說過 CNCF(Cloud Native Computing Foundation)，而 Fluentd 就是其中的一員，成為了容器圈裡多數人日誌收集處理的首選。Fluentd 的架構與 Logstash 相似，也大致分為輸入、輸出以及中間的處理層，但與之不同的是，其中間的處理層除了包括常規的 filter(解析) 以及 buffer(數據暫存) 以外，還包含了一個 routing(路由) 功能，這是 Fluentd 的一大特色。routing 功能使得 Fluentd 能夠將自己稱為一個統一的日誌管理中間層，將所有的數據輸入和輸出管理起來，按照需求將輸入的數據路由到一個或多個輸出端。這是一個非常先進的設計，其他類似的開源軟體往往要寫多份配置文件才能達到這個效果。

Fluentd 由 CRuby 實現，性能表現優良但依賴 Ruby 環境；相較於前面兩者，Fluentd 插件支持相對較少；其配置也過於複雜，使用門檻較高。

Telegraf/cAdvisor: 這兩款均是 Go 語言編寫的針對系統信息數據收集的開源工具，其側重點在 metric 收集，相較於通用的日誌收集和處理，其功能面較窄，但是性能方面均表現優異。Telegraf 配合 influxdb，可以讓你對機器各個維度的信息了如指掌；而 cAdvisor 更是 Kubernetes 的親兒子，處理容器資源信息幾無敵手。

但是這兩款工具並無意於發力通用數據收集的功能，功能上可能無法滿足一些日誌收集的場景。

了解了以上這些開源軟體的特點後，下面我們開始深入介紹構建一款數據收集工具會遇到哪些設計與挑戰，以此為你的業務量身定製。

數據收集工具設計

架構設計

主流數據收集工具的主架構基本分為 reader、parser，以及 sender 三部分，如圖 1 所示。除了這三個日誌收集常規組成部分，還應該包含可選模塊，如基於解析過後的數據轉換 (filter/transformer) 以及數據暫存管道 (Channel/Buffer)。為了儘可能復用，每個組成部分都應該是插件式的，可以編寫不同類型插件並且靈活地組裝。Channel/Buffer 部分也應該提供基於內存或者基於磁碟的選擇。

圖 1 數據收集架構設計

對於 Reader、Parser、Sender 等插件共同組裝的業務數據收集單元，我們稱之為一個運行單元 (Runner)，數據收集工具必須可以同時運行多個 Runner，且每個 Runner 可以支持更新。

更新可以通過多種方式實現，最常規的是手動更新配置然後重啟；更好的設計是支持熱更新，不需要重啟，自動識別配置文件的變化；還可以設計一個漂亮的 web 界面做配置的變更，以此引導用戶使用並解決數據收集配置複雜、用戶使用門檻高的難題。所以在整體架構之上還應該構建一個簡單的 API 層，支持 web 界面的功能。

語言選擇

數據收集屬於輕量級的 agent 服務，一般選擇的語言為 C/C++ 或者近年來特別火熱的 Go，而 Go 語言已經成為這類數據收集工具編寫的大眾選擇，如 Logstash 新開發的 beats 工具、Telegraf、cAdvisor 等等，均使用 Go 語言開發。

社區已經有很多文章描述使用 Go 語言的好處，在此就不再贅述。總體而言用 Go 語言開發門檻較低，性能優良，支持跨多種操作系統平台，部署也極為簡便。

分模塊設計

數據讀取模塊（Reader）

顧名思義，數據讀取模塊負責從不同數據源中讀取數據，設計 Reader 模塊的時候，需要支持插件式數據源接入，且將介面設計得足夠簡單，方便大家一同貢獻更多的讀取數據源驅動。

自定義數據源，最基本的只需要實現如下兩個方法即可。

從數據來源上分類，數據讀取大致可分為從文件讀取、從數據存儲服務端讀取以及從消息隊列中讀取三類。

每一類 Reader 均在發送成功後通過 SyncMeta() 函數記錄讀取的位置，保證數據不會因為 runner 意外中斷而丟失。

從文件讀取數據最為常見，針對文件的不同 rotate 方式，有不同的讀取模式，主要分為三類：

file 模式：使用 file 模式的經典日誌存儲方式類似於 nginx 的日誌 rotate 方式，日誌名稱是固定的，如，rotate 時直接 move 成新的文件如，新的數據仍然寫入到。即我們永遠只針對這一個固定名稱的文件進行收集。而檢測文件是否 rotate 的標誌是文件的 inode 號，在 windows 下則是 fd 的屬性編號。當文件 rotate 後，則從文件頭開始讀取。

dir 模式：使用 dir 模式的經典日誌存儲方式為整個文件夾下存儲單個服務的業務日誌，文件夾下的日誌通常有統一前綴，後綴為時間戳，根據日誌的大小 rotate 到新的文件。如配置的文件夾為 logdir，下面的文件為 , , , ...。每次分割後新命名文件並以時間戳為後綴，並且該文件夾下只有這一個服務。dir 模式首先會對文件夾下文件整體排序，依次讀取各個文件，讀完最後一個文件後會查找時間 (文件 ctime) 更新文件並重新排序，依次循環。dir 模式應該將多個文件數據串聯起來，即數據讀取過程中 a.log.20170621 中最後一行的下一行就是 a.log.20170622 的第一行。該模式下自然還包括諸如文件前綴匹配、特定後綴忽略等功能。

tailx 模式：以通配的路徑模式讀取，讀取所有被通配符匹配上的日誌文件，對於單個日誌文件使用 file 模式不斷追蹤日誌更新，例如匹配路徑的模式串為 , 此時會展開並匹配所有符合該表達式的文件，並持續讀取所有有數據追加的文件。每隔一定時間，重新獲取一遍模式串，添加新增的文件。

除此之外，還應支持包括多文件編碼格式支持、讀取限速等多種功能。

從數據存儲服務中讀取數據，可以採用時間戳策略，在諸如 MongoDB、MySQL 中記錄的數據，包含一個時間戳欄位，每次讀取數據均按這個時間戳欄位排序，以此獲得新增的數據或者數據更新。另一方面，需要為用戶設計類似定時器等策略，方便用戶多次運行，不斷同步收集伺服器中的數據。

從消息隊列中讀取數據，這個最為簡單，直接從消息隊列中消費數據即可。注意記錄讀取的 Offset，防止數據丟失。

解析模塊 (Parser)

解析模塊負責將數據源中讀取的數據解析到對應的欄位及類型，目前常見的解析器包括：

csv parser: 按照分隔符解析成對應欄位和類型，分隔符可以自定義，如常見的製表符 ()、空格 ()、逗號 () 等等。

json parser: 解析 json 格式的數據，json 是一種自帶欄位名稱及類型的序列化協議，解析 json 格式僅需反序列化即可。

基於正則表達式 (grok) parser: Logstash grok 解析非常強大，但是它並不指定類型，而 Telegraf 做了一個增強版的 grok 解析器，除了基本的正則表達式和欄位名稱，還能標誌數據類型，基本上能標誌數據類型的 grok 解析器已經是一個完備的數據解析器了，可以解析幾乎所有數據。當然，類型解析是相對複雜的功能，可能涉及到具體業務，如時間類型等。

raw parser: 將讀取到的一行數據作為一個欄位返回，簡單實用。

nginx/apache parser: 讀取 nginx/apache 等常見配置文件，自動生成解析的正則表達式，解析 nginx/apache 日誌。

除了以上幾種內置的解析器，同 Reader 一樣，你也需要實現自定義解析器的插件功能，而 Parser 極為簡單，只需要實現最基本的 Parse 方法即可。

每一種 Parser 都是插件式結構，可以復用並任意選擇。在不考慮解析性能的情況下，上述幾種解析器基本可以滿足所有數據解析的需求，將一行行數據解析為帶有 Schema（具備欄位名稱及類型）的數據。但是當你希望對某個欄位做操作時，純粹的解析器可能不夠用。於是作為補充，數據收集工具還需要提供 Transformer/Filter 的功能。

Transformer

Transformer 是 Parser 的補充，針對欄位進行數據變化。

舉例來說，如果你有個欄位想做字元串替換，比如將所有欄位名稱為"name"的數據中，值為"Tom"的數據改為"Tim"，那麼可以添加一個字元串替換的 Transformer，針對"name"這個欄位做替換。

又比如說，你的欄位中有個"IP"，你希望將這個 IP 解析成運營商、城市等信息，那麼你就可以添加一個 Transformer 做這個 IP 信息的轉換。

當然，Transformer 應該可以多個連接到一起連動合作。

設計 Transformer 模塊是一件有趣而富有挑戰的事情，這涉及到 Tranformer 功能多樣性帶來的 3 個問題：

多樣的功能必然涉及到多樣的配置，如何將不同的配置以優雅而統一的方式傳達到插件中？

多樣的功能也涉及到不同功能的描述，如何將功能描述以統一的形式表達給用戶，讓用戶選擇相應的配置？

如何將上述兩個問題儘可能簡單地解決，讓用戶編寫 Transformer 插件時關注儘可能少的問題？

這裡我們留個懸念，感興趣的朋友可以閱讀 logkit Transformer 相關 (https://github.com/qiniu/logkit/tree/develop/transforms) 的源碼尋求答案，筆者後續也會在 logkit 的 wiki 頁面中描述。

Channel

經過解析和變換後的數據可以認為已經處理好了，此時數據會進入待發送隊列，即 Channel 部分。Channel 的好壞決定了一個數據收集發送工具的性能及可靠程度，是數據收集工具中最具技術含量的一環。

數據收集工具，顧名思義，就是將數據收集起來，再發送到指定位置，而為了將性能最優化，我們必須把收集和發送解耦，中間提供一個緩衝帶，而 Channel 就是負責數據暫存的地方。有了 Channel，讀取和發送就解耦了，可以利用多核優勢，多線程發送數據，提高數據吞吐量。

圖 2 ft sender

一種設計思路是把整個 Channel，包括多線程發送做成一個框架，封裝成一個特殊的 sender，我們稱這個特殊的 sender 為"ft sender"。其架構如圖 2 所示，ft sender 與其他 sender 一樣也提供了 Send() 方法，解析完畢後的數據調用 Send 方法實際上就是將數據傳入到 Channel 中，然後再由 ft sender 處理多線程發送邏輯，將從隊列中取出的數據交由實際的 sender 多線程發送。

同時需要為用戶提供磁碟和內存兩大隊列方式選擇。

如果追求最高的可靠性，就使用磁碟隊列，數據會暫存到本地磁碟中，發送後自動刪除，即使突然宕機也不怕丟失數據。

如果追求更高的性能，可以使用內存隊列，其實現方式就是 Go 語言的 Channel 機制，穩定而簡單，在關停過程中也需要將 Channel 中的數據落地到磁碟，在隨後重新啟動時載入，正常使用過程中也沒有數據丟失的風險。得益於 Go 語言的同步 Channel 機制，甚至可以把內存隊列的大小設置為 0，以此實現多線程發送，這樣使用內存隊列即使宕機，也沒有了數據丟失的風險。

除了正常地作為待發送數據的等待隊列以外，Channel 還可以具有如下一些非常有趣而實用的功能：

錯誤數據篩選

並不是所有解析完畢的數據發送到服務端就一定是正確的，有時服務端指定的數據格式和解析完畢的格式存在出入，或者數據中含有一些非法字元等情況，則數據不能發送成功。此時，如果一批數據中只有一條這樣錯誤的數據，就很容易導致這一整批都失敗。

錯誤數據篩選的作用就在於，把這一整批數據中對的數據篩選出來，留下錯誤的數據，將正確的發送出去。

做法很簡單，當發送時遇到服務端返回存在格式錯誤的數據時，將這一批數據平均拆分為兩批（二分），再放入隊列，等待下次發送。再遇到錯誤時則再拆分，這樣不斷二分，直到一個批次中只有一條數據，且發送失敗，那我們就找到了這個錯誤數據，可以選擇丟棄或記錄。

藉助隊列，我們很容易就能將錯誤數據篩選出來。

包拆分（大包拆小包）

包拆分的由來是服務端不可能無限制開放每個批次數據傳輸的大小，出於伺服器性能、傳輸性能等原因，總有會有一些限制。

當一個批次的數據過大時，就會導致傳輸失敗。此時的做法與錯誤篩選的方法相似，只要將包二分即可，如果還是太大就再次二分，以此類推。

限速

限速的功能最容易理解，數據統統經過 Channel，那麼只要限制這個 Channel 傳輸介質的速度即可。例如磁碟隊列，只需要限制磁碟的 IO 讀寫速度；內存隊列則限制隊列大小以此達到限速的目的。

常見的流量控制的演算法有漏桶演算法（Leaky bucket）（https://en.wikipedia.org/wiki/Leaky_bucket）和令牌桶演算法（Token bucket）（https://en.wikipedia.org/wiki/Token_bucket）兩種，比較推薦採用令牌桶演算法實現該功能，感興趣的朋友可以閱讀一下 logkit 的 rateio 包 (https://github.com/qiniu/logkit/tree/develop/rateio)。

Sender

Sender 的主要作用是將隊列中的數據發送至 Sender 支持的各類服務，一個最基本的實現同樣應該設計得儘可能簡單，理論上僅需實現一個 Send 介面即可。

那麼實現一個發送端有哪些注意事項呢？

多線程發送：多線程發送可以充分利用 CPU 的多核能力，提升發送效率，這一點我們在架構設計中通過設計 ft sender 作為框架解決了該問題。

錯誤處理與等待：服務端偶爾出現一些異常是很正常的事情，此時就要做好不同錯誤情況的處理，不會因為某個錯誤而導致程序出錯，另外一方面，一旦發現出錯應該讓 sender 等待一定時間再發送，設定一個對後端友好的變長錯誤等待機制也非常重要。一般情況下，可以採用隨著連續錯誤出現遞增等待時間的方法，直到一個最頂峰（如），就不再增加，當服務端恢復後再取消等待。

數據壓縮發送：帶寬是非常珍貴的資源，通常服務端都會提供 gzip 壓縮的數據接收介面，而 sender 利用這些介面，將數據壓縮後發送，能節省大量帶寬成本。

帶寬限流：通常情況下數據收集工具只是機器上的一個附屬程序，主要資源如帶寬還是要預留給主服務，所以限制 sender 的帶寬用量也是非常重要的功能，限流的方法可以採用前面 Channel 一節提到的令牌桶演算法。

欄位填充 (UUID/timestamp)：通常情況下收集的數據信息可能不是完備的，需要填充一些信息進去，如全局唯一的 UUID、代表收集時間的 timestamp 等欄位，提供這些欄位自動填充的功能，有利於用戶對其數據做唯一性、時效性等方面的判斷。

欄位別名：解析後的欄位名稱中經常會出現一些特殊字元，如"$"、"@"等符號，如果發送的服務端不支持這些特殊字元，就需要提供重命名功能，將這些欄位映射到一個別的名稱。

欄位篩選：解析後的欄位數據未必都需要發送，這時如果能提供一個欄位篩選的功能，就可以方便用戶選擇去掉一些無用欄位，並節省傳輸的成本。也可以在 Transformer 中提供類似的功能，將某個欄位去掉。

類型轉換：類型轉換是一個說來簡單但是做起來非常繁瑣的事情，不只是純粹的整型轉換成浮點型，或者字元串轉成整型這麼簡單，還涉及發送到的服務端支持的一些特殊類型，如時間類型等，更多的類型轉換實際上相當於最佳實踐，能夠做好這些類型轉換，就會讓用戶體驗得到極大提升。

簡單、簡單、簡單：除了上述這些，剩下的就是儘可能的讓用戶使用簡單。假設我們要寫一個 mysql sender，mysql 的資料庫和表如果不存在，可能數據會發送失敗，那就可以考慮提前創建；又比如數據如果有更新，那麼就需要將針對這些更新的欄位去更新服務的 Schema 等等。

Metrics

除了基本的自定義的數據收集，數據收集工具作為一個機器的 agent，還可以採集機器的基本數據，例如 CPU、內存、網路等信息，通過這些信息，可以全面掌控機器的狀態。

具體的內容可以參考 logkit 文檔：Runner 之系統信息採集配置。

至此，一個完整的數據收集工具的設計要點已經介紹完畢。

我們已經開源的 logkit 正是按照這樣的設計實現的，logkit 集合了多種開源數據收集工具的優點，聚焦易用性，致力於打造產品級別的開源軟體。

數據收集工具 logkit

logkit（https://github.com/qiniu/logkit）是七牛大數據團隊開源的一個通用的日誌收集工具，可以從多種不同的數據源中採集數據，並對數據進行一系列的解析、變換、裁減，最終發送到多種不同的數據下游，其中就包括了七牛的大數據平台 Pandora。除了基本的數據收集、解析以及發送功能之外，logkit 集合了多種同類開源軟體的優勢，涵蓋了容錯、並發、熱載入、斷點續傳等高級功能，更提供了頁面方便用戶配置、監控以及管理自己的數據收集業務，是一款產品級別的開源軟體。

目前支持的數據源包括：

File Reader: 讀取文件中的日誌數據，如 nginx/apache 服務日誌文件、業務日誌等。

Elasticsearch Reader: 全量導出 Elasticsearch 中的數據。

MongoDB Reader: 同步 MongoDB 中的數據。

MySQL Reader: 同步 MySQL 中的數據。

MicroSoft SQL Server Reader: 同步 Microsoft SQL Server 中的數據。

Kafka Reader: 導出 Kafka 中的數據。

Redis Reader: 導出 Redis 中的數據。

目前支持發送到的服務包括 Pandora、ElasticSearch、InfluxDB、MongoDB 以及本地文件五種，近期還會支持發送到 Kafka 以及發送到某個 HTTP 地址。

Pandora Sender: 發送到 Pandora(七牛大數據處理平台) 服務端。

Elasticsearch Sender: 發送到 Elasticsearch 服務端。

File Sender: 發送到本地文件。

InfluxDB Sender: 發送到 InfluxDB 服務端。

MongoDB Sender: 後發送到 MongoDB 服務端。

而在這已經實現的有限的幾個發送端中，我們是這麼設計的使用場景：

如果收集數據是為了監控，那麼可以使用 InfluxDB Sender，發送到開源的 InfluxDB 服務端，實現實時數據監控。

如果收集數據是為了搜索查看，那麼可以使用 Elasticsearch Sender，發送到開源的 Elasticsearch 服務端，進行日誌查詢。

如果收集數據是為了計量統計或者其他一些涵蓋複雜的增刪改查需求的場景，那麼就可以使用 MongoDB Sender，在本地對數據進行聚合，再發送到開源的 MongoDB 服務中。

而發送到七牛的 Pandora 服務中，不僅能涵蓋上述全部場景，還提供了大量可以快速發掘數據價值的實際應用場景的使用模板，同時還可以利用七牛成本較低的雲存儲服務對數據進行持久備份。

目前 logkit 支持的收集端和發送端並不多，非常歡迎大家來貢獻更多的收集 / 發送端。

量身定製

再回過頭來聊聊量身定製的話題，本文描述了一個數據收集工具打造的完整過程，我們深知量身定製一個數據收集工具有多麼重要，而量身定製也是 logkit 的一大特點。logkit 架構中的每個組成部分（Reader、Parser、Sender、Transformer、Channel 等）都是一個 GO 語言的 package，你完全可以用 logkit 中的包，自己寫主函數，從而定製一個專屬的收集工具。我們提供了代碼案例 (https://github.com/qiniu/logkit/tree/develop/samples) ，方便你親自實踐。

優勢

總體而言，logkit 有如下優勢：

GO 語言編寫，性能優良，資源消耗低，跨平台支持。

Web 支持，提供頁面對數據收集、解析、發送過程可視化

插件式架構，擴展性強，使用靈活，易於復用。

定製化能力強，可以僅使用部分 logkit 包，以此定製專屬收集工具。

配置簡單，易於上手，可通過頁面進行操作管理

原生中文支持，沒有漢化煩惱

功能全面，涵蓋了包括 grok 解析、metric 收集、欄位變化 (transform) 在內的多種開源軟體特點

生態全面，數據發送到七牛的 Pandora 大數據平台支持包括時序資料庫、日誌檢索以及壓縮永久存儲等多種數據落地方案。

下面讓我們來實踐一下，看看 logkit 在實戰中是什麼樣子。

logkit 實戰