導讀：資料總線DBus的總體架構中主要包括六大子產品，分别是：日志抓取子產品、增量轉換子產品、全量抽取程式、日志算子處理子產品、心跳監控子產品、Web管理子產品。六大子產品各自的功能互相連接配接，構成DBus的工作原理：通過讀取RDBMS增量日志的方式來實時擷取增量資料日志（支援全量拉取）；基于Logstash，flume，filebeat等抓取工具來實時獲得資料，以可視化的方式對資料進行結構化輸出。本文主要介紹的是DBus中基于可視化配置的日志結構化轉換實作的部分。
一、結構化日志的原理
1.1 源端日志抓取
DBus可以對接多種log資料源，例如：Logstash、Flume、Filebeat等。上述元件都是業界比較流行的日志抓取工具，一方面便于使用者和業界統一标準，友善使用者技術方案的整合；另一方面也避免了無謂的重複造輪子。抓取的資料我們稱為原始資料日志（raw data log），由抓取元件将其寫入Kafka中，等待DBus後續處理。
1.2 可視化配置規則，使日志結構化
使用者可自定義配置日志源和目标端。同一個日志源的資料可以輸出到多個目标端。每一條“日志源-目标端”線，使用者可以根據自己的需要來配置相應的過濾規則。經過規則算子處理後的日志是結構化的，即：有schema限制，類似于資料庫中的表。
1.3 規則算子
DBus設計了豐富易用的算子，用于對資料進行定制化操作。使用者對資料的處理可分為多個步驟進行，每個步驟的資料處理結果可即時檢視、驗證；并且可重複使用不同算子，直到轉換、裁剪出自己需要的資料。
1.4 執行引擎
将配置好的規則算子組應用到執行引擎中，對目标日志資料進行預處理，形成結構化資料，輸出到Kafka，供下遊資料使用方使用。系統流程圖如下所示：
根據DBus log設計原則，同一條原始日志，可以被提取到一個或多個表中。每個表是結構化的，滿足相同的schema限制。
- 每個表是一個規則算子組的集合，每個表可以擁有1個或多個規則算子組；
- 每個規則算子組，由一組規則算子組合而成，每個算子具有獨立性；
對于任意一條原始資料日志（raw data log），它應該屬于哪張表呢？
假如使用者定義了若幹張邏輯表（T1,T2…），用于抽取不同類型的日志，那麼，每條日志需要與規則算子組進行比對：
- 進入某張表T1的所有規則算子組的執行過程
- 符合條件的進入規則算子組，并且被執行引擎轉換為結構化的表資料
- 不符合提取條件的日志嘗試下一個規則算子組
- 對于T1的所有規則算子組，如果都不滿足要求，則進入下一張表T2的執行過程，以此類推
- 如果該條日志不符合任何一張表的過濾規則，則進入_unknown_table_表
例如，對于同一條應用日志，其可能屬于不止一個規則組或Table，而在我們定義的規則組或Table中，隻要其滿足過濾條件，該應用日志就可以被規則組提取，即保證了同一條應用日志可以同屬于不同的規則組或Table。
規則算子是對資料進行過濾、加工、轉換的基本單元。常見的規則算子如上圖所示。
算子之間具有獨立性，算子之間可以任意組合使用，進而可以實作許多複雜的、進階的功能，通過對算子進行疊代使用，最終可以實作對任意資料進行加工的目的。使用者可以開發自定義算子，算子的開發非常容易，使用者隻要遵循基本接口原則，就可以開發任意的算子。
二、DBus日志處理執行個體
以DBus叢集環境為例，DBus叢集中有兩台機器（即master-slave）部署了心跳程式，用于監控、統計、預警等，心跳程式會産生一些應用日志，這些應用日志中包含各類事件資訊，假如我們想要對這些日志進行分類處理并結構化到資料庫中，我們就可以采用DBus log程式對日志進行處理。
DBus可以接入多種資料源（Logstash、Flume、Filebeat等），此處以Logstash為例來說明如何接入DBus的監控和報警日志資料。
由于在dbus-n2和dbus-n3兩台機器上分别存在監控和預警日志，為此我們分别在兩台機器上部署了Logstash程式。心跳資料由Logstash自帶的心跳插件産生，其作用是便于DBus對資料進行統計和輸出，以及對源端日志抽取端（此處為Logstash）進行預警（對于Flume和Filebeat來說，因為它們沒有心跳插件，是以需要額外為其定時産生心跳資料）。Logstash程式寫入到Kafka中的資料中既有普通格式的資料，同時也有心跳資料。
這裡不隻是局限于2台部署有Logstash程式的機器，DBus對Logstash數量不做限制，比如應用日志分布在幾十上百台機器上，隻需要在每台機器上部署Logstash程式，并将資料統一抽取到同一個Kafka Topic中，DBus就能夠對所有主機的資料進行資料處理、監控、預警、統計等。
2.1 啟動Logstash
在啟動Logstash程式後，我們就可以從topic : heartbeat_log_logstash中讀取資料，資料樣例如下：
1)心跳資料
2)普通日志資料
2.2 配置規則
接下來，我們隻需要在DBus Web中配置相應的規則就可以對資料進行處理了。
首先建立一個邏輯表sink_info_table，該表用來抽取sink事件的日志資訊，然後配置該表的規則組（一個或多個，但所有的規則組過濾後的資料需要滿足相同schema特性），heartbeat_log_logstash作為原始資料topic，我們可以實時的對資料進行可視化操作配置（所見即所得，即席驗證）。
1)讀取原始資料日志
可以看到由Logstash預先提取已經包含了log4j的基本資訊，例如path、@timestamp、level等。但是資料日志的詳細資訊在字段log中。由于不同的資料日志輸出是不一樣的，是以可以看到log列資料是不同的。
2)提取感興趣的列
假如我們對timestamp、log 等原始資訊感興趣，那麼可以添加一個toIndex算子，來提取這些字段：
這裡需要指出，我們考慮使用數組下标方式，是有原因的：
- 并不是所有列本身自帶列名（例如flume抽取的原始資料，或者split算子處理後的資料列）；
-
下标方式可以使用數組方式指定列（類似python方式， 例如：1:3表示1，2列）；
是以後續操作全部基于數組下标方式通路。
執行規則，就可以看到被提取後的字段情況：
3)過濾需要的資料
在這個例子中，我們隻對含有“Sink to influxdb OK!”的資料感興趣。是以添加一個filter算子，提取第7列中包含”Sink to influxdb OK!”内容的行資料：
執行後，隻有符合條件的日志行資料才會存在。
4)對特定列進行提取
添加一個select算子，我們對第1和3列的内容感興趣，是以對這兩列進行提取。
執行select算子，資料中就會隻含有第1和3列了。
5)以正規表達式的方式處理資料
我們想從第1列的資料中提取符合特定正規表達式的值，使用regexExtract算子對資料進行過濾。正規表達式如下：http_code=(d*).*type=(.*),ds=(.*),schema=(.*),table=(.*)s.*errorCount=(d*)，使用者可以寫自定義的正規表達式。
執行後，就會擷取正規表達式執行後的資料。
6)選擇輸出列
最後我們把感興趣的列進行輸出，使用saveAs算子， 指定列名和類型，友善于儲存在關系型資料庫中。
執行saveAs算子後，這就是處理好的最終輸出資料樣本。
2.3 檢視結構化輸出結果
儲存上一步配置好的規則組，日志資料經過DBus執行算子引擎，就可以生成相應的結構化資料了。目前根據項目實際，DBus輸出的資料是UMS格式，如果不想使用UMS，可以經過簡單的開發，實作定制化。
注：UMS是DBus定義并使用的、通用的資料交換格式，是标準的JSON。其中同時包含了schema和資料資訊。更多UMS介紹請參考DBus開源項目首頁的介紹。開源位址：
https://github.com/bridata/dbus以下是測試案例，輸出的結構化UMS資料的樣例：
2.4 日志監控
為了便于掌握資料抽取、規則比對、監控預警等情況，我們提供了日志資料抽取的可視化實時監控界面，如下圖所示，可随時了解以下資訊：
- 實時資料條數
- 錯誤條數情況（錯誤條數是指：執行算子時出現錯誤的情況，幫助發現算子與資料是否比對，用于修改算子，DBus同時也提供了日志回讀的功能，以免丢失部分資料）
- 資料延時情況
- 日志抽取端是否正常
監控資訊中包含了來自叢集内各台主機的監控資訊，以主機IP（或域名）對資料分别進行監控、統計和預警等。
監控中還有一張表叫做_unkown_table_ 表明所有沒有被比對上的資料條數。例如：Logstash抓取的日志中有5種不同僚件的日志資料，我們隻捕獲了其中3種事件，其它沒有被比對上的資料，全部在_unkown_table_計數中。
DBus同樣可以接入Flume、Filebeat、UMS等資料源，隻需要稍作配置，就可以實作類似于對Logstash資料源同樣的處理效果，更多關于DBus對log的處理說明，請參考：
- https://bridata.github.io/DBus/install-logstash-source.html
- https://bridata.github.io/DBus/install-flume-source.html
- https://bridata.github.io/DBus/install-filebeat-source.html
應用日志經過DBus處理後，将原始資料日志轉換為了結構化資料，輸出到Kafka中提供給下遊資料使用方進行使用，比如通過Wormhole将資料落入資料庫等。具體如何将DBus與Wormhole結合起來使用，請參考：
如何設計實時資料平台（技術篇）。
作者：仲振林