Logstash

之前用的Logstash快忘了，好記性不如爛筆頭，好好總結一下。
Logstash由Java(Core)+Ruby(Plugin)語言編寫，是一個開源的日誌收集、處理、轉發工具。
Input產生事件，Filter修改事件，Output將事件傳送到其他地方。
注:經常看到的編解碼器codec也是過濾器，可以作為輸入或輸出的一部分,對流進行編解碼。

簡單部署

#下載解壓
[[email protected] ~/software]$ wget https://artifacts.elastic.co/downloads/logstash/logstash-6.3.2
 
.tar.gz

[[email protected] ~/software]$ tar -zxvf logstash-6.3.2.tar.gz

#自檢-接收標準輸入並輸出到控制檯
[[email protected] ~/software/logstash-6.3.2]$ bin/logstash -e 'input { stdin { } } output { stdout {} }'
輸入 Hello
輸出
    {
          "@version" => "1",
           "message" => "Hello",
        "@timestamp"
 
 => 2018-08-11T11:49:31.059Z,
              "host" => "localhost.local"
    }

Mysql增量同步至 Kafka/Hdfs/Elasticsearch

Mysql 測試庫表

生產中一個庫同一類資料一般會有多張分表如16張分表,256張分表。

#建庫 user_logs
create database user_logs;

#建表-使用者瀏覽日誌表: user_browse_0_logs,user_browse_1_logs 如
CREATE TABLE `user_browse_0_logs`
 
 (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `product_id` int(11) NOT NULL,
  `event_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  `uid` varchar(10) NOT NULL,
  PRIMARY KEY (`id`),
  KEY `event_time` (`event_time`) USING BTREE,
  KEY `uid_event_time` (`uid`,`event_time`) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='使用者瀏覽日誌表';

Logstash 配置

基礎配置

#配置目錄
[wangpei@localhost ~/software/logstash-6.3.2]$ mkdir db_test

#拷貝預設配置
[wangpei@localhost ~/software/logstash-6.3.2]$ cp config/* db_test/

#db_test/jvm.options
#jvm配置 
-Xms1g
-Xmx1g

#db_test/logstash.yml
#持久化目錄 
#logstash和plugin任何需要持久化的資料都會放在這個目錄
path.data: /Users/wangpei/software/logstash-6.3.2/data/user_browse_logs/data

#queue  
#預設memory,logstash程序重啟會丟資料
queue.type: persisted

#併發度
pipeline.workers: 2

#每個執行緒從輸入收集的最大數量
pipeline.batch.size: 125

Input-Filter-Output配置

[wangpei@localhost ~/software/logstash-6.3.2]$ vim db_test/user_browse_logs.conf
#輸入配置 jdbc外掛
input {
  jdbc {
    #jdbc驅動
    jdbc_driver_library => "/Users/wangpei/software/logstash-6.3.2/mysql-connector-java-5.1.38.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    #jdbc連線
    jdbc_connection_string => "jdbc:mysql://node2:3306/user_logs"
    jdbc_user => "root"
    jdbc_password => "111"
    #任務排程
    schedule => "* * * * *"
    #上次執行的sql_last_value儲存路徑
    #啟動前需要確保該檔案已存在
    last_run_metadata_path => "/Users/wangpei/software/logstash-6.3.2/data/test/metadata/user_browse_0_logs"
    #sql語句
    #按主鍵自增id 增量同步
    #input在拿到欄位為日期型別的資料時,一定會減8小時並帶上時區資訊
    #如原始mysql event_time欄位為日期型別(這裡是timestamp型別)的資料2018-08-12 14:38:24 =>經過input後=> 2018-08-12T14:38:24.000Z
    #不想讓input修改原始資料,這裡直接把event_time轉成成字串型別。
    statement => "SELECT id,product_id,DATE_FORMAT(event_time, '%Y-%m-%d %H:%i:%s') as event_time,uid FROM user_browse_0_logs WHERE id > :sql_last_value"
    #設定成true 用tracking_column列的值去更新上次記錄的sql_last_value值
    use_column_value => true
    #追蹤的列
    tracking_column => "id"
    #追蹤的列的資料型別 支援["numeric", "timestamp"]
    tracking_column_type => "numeric"
    #外掛id 便於監控
    id => "user_browse_logs_input_jdbc"
    #新增欄位 表名 便於後期DQC 如mysql 每個分表的資料和HDFS每個分割槽資料count、sum驗證
    add_field => {"table_name" => "user_browse_0_logs"}
    #input event輸出格式 每個event轉換成json+換行符格式
    codec => json_lines {charset => "UTF-8"}
  }
}

#Filter配置
filter {
  #預設會對每條資料加上@timestamp
  #不指定@timestamp欄位來源,預設會以logstash伺服器收到資料時間為準
  date {
        #匹配到這種格式就會替換成目標欄位,目標欄位預設是@timestamp
        #貌似不管怎麼設定,Logstash中 @timestamp永遠是0時區
        #@timestamp 0時區,資料寫入ES，@timestamp 在Kibana上顯示時會根據瀏覽器市區自動調整 如自動+8
        match => ["event_time" , "yyyy-MM-dd HH:mm:ss"]
        #timezone => "Asia/Shanghai"
    #target => "@timestamp"
    }
  #增加dt欄位,該欄位來源於event_time,用於hdfs按天分割槽、ES按天索引
  #總結:
  #假設源資料是北京時間
  #1)資料同步到ES:(需要兼顧kibana各種作圖、Java/Python API查詢)
  #   A、按北京時間建索引如下dt
  #   B、kibana視覺化,時間選擇用@timestamp時間欄位,避免用event_time時間欄位做不了Date Histogram問題。
  #   C、ES API查詢,時間選擇用event_time時間欄位更方便。
  #2)資料同步到HDFS: 按北京時間分割槽 如下dt即可
  ruby {code => "event.set('dt', event.get('event_time')[0,10].gsub('-',''));"}
}

#輸出配置
output {
  #輸出到控制檯
  stdout { codec => rubydebug}
  #輸出到kafka
  kafka {
    #kafka brokers
    bootstrap_servers => "localhost:9092"
    #topic
    topic_id => "testTopic3"
    #ack
    acks => "1"
    #輸出格式
    codec => json
    #id
    id => "user_browse_logs_ouput_kafka"
  }
  #輸出到HDFS
  #需要開啟HDFS 的webhdfs
  webhdfs {
    codec => json_lines {charset => "UTF-8"}
    #namenode
    host => 'node1'
    port => 50070
    #standby namenode
    standby_host => 'node4'
    standby_port => 50070
    #user
    user => 'root'
    #user需要對path有讀寫許可權
    path => '/data/user_logs/user_browse_logs/%{dt}/%{table_name}.log'
    #攢到多少條event時向hdfs寫一次資料
    flush_size => 3
    #間隔多久向hdfs寫一次資料 單位:秒
    idle_flush_time => 2
  }
  #輸出到elasticsearch
  elasticsearch {
         #es節點列表
         hosts => ["localhost:9200"]
         #索引使用的模板
         #模板中定義了mapping(日期Format,某個欄位使用的分詞器)和setting
     template_name => "template.user_browse_logs"
         #index
         #資料是北京時間,這個按北京時間建索引
         index => "user_browse_logs.%{dt}"
     #type
         document_type =>"browse_logs"
         #許可權
         user => "logstash_input"
         password => "logstash_input"
         codec => json {charset => "UTF-8"}
    }
}

啟動方式

#直接啟動===> 一個input-filter-output配置獨享一個logstash程序
[[email protected] ~/software/logstash-6.3.2]$ bin/logstash --path.settings db_test/  -f db_test/user_browse_logs.conf --config.reload.automatic

#pipeline方式啟動===> 多個input-filter-output配置共享一個logstash程序
[[email protected] ~/software/logstash-6.3.2]$ bin/logstash --path.settings db_test/  --config.reload.automatic
需配置db_test/pipelines.yml
配置多個pipeline。每個pipeline可對應一個表(input-filter-output)。
- pipeline.id: user_logs.user_browse_logs
  queue.type: persisted
  pipeline.workers: 1
  queue.max_bytes: 32mb
  path.config: "/Users/wangpei/software/logstash-6.3.2/db_test/user_browse_logs.conf"

- pipeline.id: user_logs.user_pay_logs
  queue.type: persisted
  pipeline.workers: 1
  queue.max_bytes: 32mb
  path.config: "/Users/wangpei/software/logstash-6.3.2/db_test/user_pay_logs.conf"
  ...

結果

mysql插入資料

insert into user_browse_0_logs(product_id,uid) values(1,"0");
insert into user_browse_0_logs(product_id,uid) values(2,"0");
insert into user_browse_0_logs(product_id,uid) values(3,"0");

mysql> select * from user_browse_0_logs;
+----+------------+---------------------+-----+
| id | product_id | event_time          | uid |
+----+------------+---------------------+-----+
|  1 |          1 | 2018-08-12 20:18:31 | 0   |
|  2 |          2 | 2018-08-12 20:18:31 | 0   |
|  3 |          3 | 2018-08-12 20:18:32 | 0   |
+----+------------+---------------------+-----+

kafka收到的資料

[wangpei@localhost ~/software/logstash-6.2.4]$ kafka-console-consumer --bootstrap-server localhost:9092 --topic testTopic3

{"@version":"1","dt":"20180812","product_id":3,"uid":"0","id":3,"table_name":"user_browse_0_logs","event_time":"2018-08-12 20:18:32","@timestamp":"2018-08-12T12:18:32.000Z"}
{"@version":"1","dt":"20180812","product_id":1,"uid":"0","id":1,"table_name":"user_browse_0_logs","event_time":"2018-08-12 20:18:31","@timestamp":"2018-08-12T12:18:31.000Z"}
{"@version":"1","dt":"20180812","product_id":2,"uid":"0","id":2,"table_name":"user_browse_0_logs","event_time":"2018-08-12 20:18:31","@timestamp":"2018-08-12T12:18:31.000Z"}

hdfs收到的資料

#分割槽:20180812
#分割槽下的檔案:user_browse_0_logs.log user_browse_0_logs是表名
#該路徑來自:output webhdfs path引數 path => '/data/user_logs/user_browse_logs/%{dt}/%{table_name}.log'

[root@node2 /root]# hdfs dfs -cat /data/user_logs/user_browse_logs/20180812/user_browse_0_logs.log
{"@version":"1","dt":"20180812","product_id":1,"uid":"0","id":1,"table_name":"user_browse_0_logs","event_time":"2018-08-12 20:18:31","@timestamp":"2018-08-12T12:18:31.000Z"}
{"@version":"1","dt":"20180812","product_id":2,"uid":"0","id":2,"table_name":"user_browse_0_logs","event_time":"2018-08-12 20:18:31","@timestamp":"2018-08-12T12:18:31.000Z"}
{"@version":"1","dt":"20180812","product_id":3,"uid":"0","id":3,"table_name":"user_browse_0_logs","event_time":"2018-08-12 20:18:32","@timestamp":"2018-08-12T12:18:32.000Z"}

es收到的資料

檢視jdbc偏移量記錄檔案

#該路徑來自input jdbc外掛 last_run_metadata_path引數
last_run_metadata_path => "/Users/wangpei/software/logstash-6.3.2/data/test/metadata/user_browse_0_logs"

#此時id的偏移量是3,下次取id>3的資料。這樣就實現了增量同步
[wangpei@localhost ~/software/logstash-6.3.2]$ cat /Users/wangpei/software/logstash-6.3.2/data/test/metadata/user_browse_0_logs
--- 3

檢視logstash持久化的資料

#持久化目錄下包含2個目錄,一個檔案
#該目錄來自於logstash.yml中 path.data: /Users/wangpei/software/logstash-6.3.2/data/user_browse_logs/data

[wangpei@localhost ~/software/logstash-6.3.2]$ ls -l /Users/wangpei/software/logstash-6.3.2/data/user_browse_logs/data
total 8
drwxr-xr-x  2 wangpei  staff   68  8 12 19:38 dead_letter_queue
drwxr-xr-x  4 wangpei  staff  136  8 12 19:57 queue
-rw-r--r--  1 wangpei  staff   36  8 12 19:38 uuid

#uuid:logstash agent uuid
#queue:input → queue → filter + output。queue配置為persisted時,input會把資料先寫入此持久化目錄,然後再由filter + output處理。
#dead_letter_queue:DLQ佇列。死信佇列儲存響應程式碼為400或404的情況，兩者都表示無法重試的事件。
如資料input=>filter=>output(es)，
一條資料出現mapping error，此時如果在logstash.yml中配置了dead_letter_queue.enable: true,
就會將這條資料放在DLQ對應的持久化目錄中。然後可以通過DLQ(Dead Letter Queues) Input => Filter(再處理)=>Output ES。

附:Kafka Input

input{
    kafka{
        codec => json {charset => "UTF-8"}
        #消費者組
        group_id => "test_topic_consumer1"
        #從最新開始消費
        auto_offset_reset => "latest"
        #消費的topics
        topics => ["test_topic"]
        #bootstrap server
        bootstrap_servers => "localhost:9092"
        #注意:執行緒數最好與partition數目一樣多,超過partition數量的執行緒會閒置。
        #kafka topic 一個partition同一時刻只能有一個執行緒去消費。
        consumer_threads =>3
    }
}

Logstash監控

可通過X-Pack以UI的方式檢視監控資訊，也可通過監控API檢視。
監控API主要有四類:節點資訊、外掛資訊、節點狀態統計、熱執行緒統計。
- 節點資訊

介面返回os、jvm、pipelines配置資訊 如jvm最大最小堆、pipelines worker數。
http://localhost:9600/_node/<os/pipelines/jvm>?pretty 

• 外掛資訊

列出所有可用外掛。
http://localhost:9600/_node/plugins?pretty

• 節點狀態統計

http://localhost:9600/_node/stats/<type>?pretty
type=jvm 獲取jvm統計資訊。如jvm當前執行緒數、執行緒峰值、記憶體各區(young、old、survivor)記憶體使用情況、gc次數、gc時間
type=process 獲取程序統計資訊。如記憶體消耗和CPU使用情況
type=events 獲取event相關統計資訊。如輸入、filter、輸出中的event數目
type=pipelines 獲取每個pipeline統計資訊、

• 熱執行緒統計

獲取當前熱執行緒。熱執行緒是一個Java執行緒，具有較高的CPU使用率並且執行時間超過正常時間。
http://localhost:9600/_node/hot_threads?pretty

注意

• 一個input多個output

如一個jdbc,多個output(hdfs、kafka、es)，kafka和es連線不上,input會繼續輸入,偏移量如自增id會繼續增加，配置了queue.type: persisted,output plugin kafka/es正常後,會重發歷史資料到kafka/es。原因:input在確認資料發往queue後就返回了。

• jdbc全量匯入

去掉jdbc sql語句中where條件即可。

• jdbc增量匯入給從庫同步留下時間

#如果logstash基於從庫按自增id做增量同步，由於事務提交順序並不一定是按id從小打到的順序，id雖然是自增id,但id=100的資料可能先於id=96的資料入庫。
#如若此時logstash同步了id=100的資料(id=96的資料還未入庫)，那sql_last_value就一定會>=100，下次再導的時候，只會取id>sql_last_value的資料，就會丟失了id=96這條資料。
#解決方式：給從庫同步留下時間，最大隻導1分鐘之前的資料(從庫同步延遲在毫秒級別)，即等id=96也入庫後再導。
#實踐中，注意從庫的延遲。如果從庫延遲較大，也應適當增加等待間隔。
#statement改為:
statement => "SELECT id,product_id,DATE_FORMAT(event_time, '%Y-%m-%d %H:%i:%s') as event_time,uid FROM user_browse_0_logs WHERE id > :sql_last_value and event_time < date_sub(now(),interval 1 minute)"