Fluentd插件

產品介紹

該插件是基于Fluentd開發的輸出插件，主要是將采集到的數據寫入DataHub。該插件遵守Fluentd輸出插件開發規范，安裝方便，可以很方便地將采集得到的數據寫到DataHub。

產品安裝

通過Ruby gem安裝

注意：RubyGem源建議更改為 https://gems.ruby-china.com

gem install fluent-plugin-datahub

本地安裝

1. 當前Fluentd僅支持Linux環境, 同時要求用戶安裝Ruby。

2. 目前支持兩種安裝模式，對于沒有安裝過fluentd的客戶我們提供了一鍵fluent+datahub全部安裝模式，對于曾經安裝了fluentd的客戶我們提供datahub寫入插件單獨安裝模式

1) 一鍵安裝：如果之前沒安裝過fluentd，請點擊下載Fluentd完全安裝包。注意，完全安裝包提供的fluentd是fluentd-0.12.25.gem的版本。

$ tar -xzvf fluentd-with-datahub-0.12.25.tar.gz
$ cd fluentd-with-datahub
$ sudo sh install.sh

2) 單獨安裝 如果之前安裝過fluentd，請點此下載fluentd datahub插件包, 使用gem命令安裝datahub插件。

$ sudo gem install --local fluent-plugin-datahub-0.12.25.gem

使用案例

案例一：CSV文件上傳

下面以增量的CSV文件為例，說明下如何使用Fluentd將增量的CSV文件準實時上傳到DataHub數據。CSV文件的格式如下所示：

0,qe614c760fuk8judu01tn5x055rpt1,true,100.1,14321111111
1,znv1py74o8ynn87k66o32ao4x875wi,true,100.1,14321111111
2,7nm0mtpgo1q0ubuljjjx9b000ybltl,true,100.1,14321111111
3,10t0n6pvonnan16279w848ukko5f6l,true,100.1,14321111111
4,0ub584kw88s6dczd0mta7itmta10jo,true,100.1,14321111111
5,1ltfpf0jt7fhvf0oy4lo8m3z62c940,true,100.1,14321111111
6,zpqsfxqy9379lmcehd7q8kftntrozb,true,100.1,14321111111
7,ce1ga9aln346xcj761c3iytshyzuxg,true,100.1,14321111111
8,k5j2id9a0ko90cykl40s6ojq6gruyi,true,100.1,14321111111
9,ns2zcx9bdip5y0aqd1tdicf7bkdmsm,true,100.1,14321111111
10,54rs9cm1xau2fk66pzyz62tf9tsse4,true,100.1,14321111111

上述CSV文件中每行一條Record，按照區分字段。保存在本地路徑/temp/test.csv中。DataHub Topic格式如下：

使用如下Fluentd的配置，配置文件地址在 ${CONFIG_HOME}/fluentd_test.conf：

<source>
  @type tail
  path 你的文件路徑
  tag test1
  format csv
  keys id,name,gender,salary,my_time
</source>
<match test1>
  @type datahub
  access_id your_app_id
  access_key your_app_key
  endpoint http://ip:port
  project_name test_project
  topic_name fluentd_performance_test_1
  column_names ["id", "name", "gender", "salary", "my_time"]
  flush_interval 1s
  buffer_chunk_limit 3m
  buffer_queue_limit 128
  dirty_data_continue true
  dirty_data_file 臟數據記錄文件路徑
  retry_times 3
  put_data_batch_size 1000
</match>

使用如下命令啟動Fluentd，即可完成CSV文件數據采集進入DataHub：

${FLUENTD_HOME}/fluentd-with-dataHub/bin/fluentd -c ${CONFIG_HOME}/fluentd_test.conf

案例二：Log4J日志采集

Log4j的日志格式如下：

11:48:43.439 [qtp1847995714-17] INFO  AuditInterceptor - [c2un5sh7cu52ek6am1ui1m5h] end /web/v1/project/tefe4mfurtix9kwwyrvfqd0m/node/0m0169kapshvgc3ujskwkk8g/health GET, 4061 ms

使用如下Fluentd配置：

 <source>
   @type tail
   path bayes.log
   tag test
   format /(?<request_time>\d\d:\d\d:\d\d.\d+)\s+\[(?<thread_id>[\w\-]+)\]\s+(?<log_level>\w+)\s+(?<class>\w+)\s+-\s+\[(?<request_id>\w+)\]\s+(?<detail>.+)/
 </source>
 <match test>
   @type datahub
   access_id your_access_id
   access_key your_access_key
   endpoint http://ip:port
   project_name test_project
   topic_name dataHub_fluentd_out_1
   column_names ["thread_id", "log_level", "class"]
 </match>

使用該配置啟動即可完成log4j日志采集進入DataHub的功能。

配置參數

讀插件配置
tag test1                          : 指定路由, 和<match>會進行路由正則匹配
format csv                         : 數據按照csv方式采集
keys id,name,gender,salary,my_time : 指定需要采集的列名，必須和目的DataHub表的列名一致

寫插件配置
shard_id 0               : 指定shard_id寫入，默認round-robin方式寫入
shard_keys ["id"]         : 指定用作分區key，用key值hash后作為寫入shard的索引
flush_interval 1         : fluentd 每一秒鐘至少寫一次, 默認60s
buffer_chunk_limit 3m    : 塊大小，支持“k”（KB）,“m”（MB）單位，建議值3m
buffer_queue_limit 128   : 塊隊列大小，此值與buffer_chunk_limit共同決定整個緩沖區大小
put_data_batch_size 1000 : 每1000條record寫一次DataHub
retry_times 3            : 重試次數
retry_interval  3        : 重試間隔(單位:s)
dirty_data_continue true : 遇到增數據是否繼續，若為true 遇到臟數據會重試，重試次數用完，會將臟數據寫入臟數據文件
dirty_data_file /xxx/yyy : 指定臟數據文件的位置
column_names ["id"]      : 指定需要采集的列

性能測試

性能測試環境：Fluentd運行環境為2核4G，操作系統為Linux；性能結果如下：從本次DataHub插件性能測試數據中可以看到：

1. 針對單條512B的數據，寫入速度極本保持在 2800record/s 左右。

2. 隨著put_data_batch_size的增加速度略有提升，但效果不大。

3. 對于單條100K的數據， put_data_batch_size只有100的時候可以正常work，500和1000都不可用：因為一次寫入數據過大，已經大于50m。

4. 總的平均寫入速度（MB/S）保持在 3MB/S。

   FAQ

   Q: 關于Fluentd，如何編寫Format的正則表達式？A: format正則表達式編輯器