logstash_output_kafka:Mysql同步Kafka深刻詳解

0、題記

實際業務場景中，會遇到基礎數據存在Mysql中，實時寫入數據量比較大的情景。遷移至kafka是一種比較好的業務選型方案。

而mysql寫入kafka的選型方案有：
方案一：logstash_output_kafka 插件。
方案二：kafka_connector。
方案三：debezium 插件。
方案四：flume。
方案五：其餘相似方案。
其中：debezium和flume是基於mysql binlog實現的。

若是須要同步歷史全量數據+實時更新數據，建議使用logstash。

一、logstash同步原理

經常使用的logstash的插件是：logstash_input_jdbc實現關係型數據庫到Elasticsearch等的同步。

實際上，核心logstash的同步原理的掌握，有助於你們理解相似的各類庫之間的同步。

logstash核心原理：輸入生成事件，過濾器修改它們，輸出將它們發送到其餘地方。

logstash核心三部分組成：input、filter、output。

input { }
filter { }
output { }

1.1 input輸入

包含但遠不限於：

1. jdbc：關係型數據庫：mysql、oracle等。
2. file：從文件系統上的文件讀取。
3. syslog：在已知端口514上偵聽syslog消息。
4. redis：redis消息。beats：處理 Beats發送的事件。
5. kafka：kafka實時數據流。

1.2 filter過濾器

過濾器是Logstash管道中的中間處理設備。您能夠將過濾器與條件組合，以便在事件知足特定條件時對其執行操做。

能夠把它比做數據處理的ETL環節。

一些有用的過濾包括：

1. grok：解析並構造任意文本。Grok是目前Logstash中將非結構化日誌數據解析爲結構化和可查詢內容的最佳方式。有了內置於Logstash的120種模式，您極可能會找到知足您需求的模式！
2. mutate：對事件字段執行常規轉換。您能夠重命名，刪除，替換和修改事件中的字段。
3. drop：徹底刪除事件，例如調試事件。
4. clone：製做事件的副本，可能添加或刪除字段。
5. geoip：添加有關IP地址的地理位置的信息。

1.3 output輸出

輸出是Logstash管道的最後階段。一些經常使用的輸出包括：

elasticsearch：將事件數據發送到Elasticsearch。
file：將事件數據寫入磁盤上的文件。
kafka：將事件寫入Kafka。
詳細的filter demo參考：http://t.cn/EaAt4zP

二、同步Mysql到kafka配置參考

input {
    jdbc {
      jdbc_connection_string => "jdbc:mysql://192.168.1.12:3306/news_base"
      jdbc_user => "root"
      jdbc_password => "xxxxxxx"
      jdbc_driver_library => "/home/logstash-6.4.0/lib/mysql-connector-java-5.1.47.jar"
      jdbc_driver_class => "com.mysql.jdbc.Driver"
      #schedule => "* * * * *"
      statement => "SELECT * from news_info WHERE id > :sql_last_value  order by id"
      use_column_value => true
      tracking_column => "id"        
      tracking_column_type => "numeric"
      record_last_run => true
      last_run_metadata_path => "/home/logstash-6.4.0/sync_data/news_last_run"    

    }

}

filter {
   ruby{
        code => "event.set('gather_time_unix',event.get('gather_time').to_i*1000)"
    }
    ruby{
        code => "event.set('publish_time_unix',event.get('publish_time').to_i*1000)"
    }
  mutate {
    remove_field => [ "@version" ]
    remove_field => [ "@timestamp" ]
    remove_field => [ "gather_time" ]
    remove_field => [ "publish_time" ]
  }
}

 output {
      kafka {
            bootstrap_servers => "192.168.1.13:9092"
            codec => json_lines
            topic_id => "mytopic"

    }
    file {
            codec => json_lines
            path => "/tmp/output_a.log"
    }
 }

以上內容不復雜，不作細講。

注意：
Mysql藉助logstash同步後，日期類型格式：「2019-04-20 13:55:53」已經被識別爲日期格式。

code =>
"event.set('gather_time_unix',event.get('gather_time').to_i*1000)"，

是將Mysql中的時間格式轉化爲時間戳格式。

三、坑總結

3.1 坑1字段大小寫問題

from星友：使用logstash同步mysql數據的，由於在jdbc.conf裏面沒有添加 lowercase_column_names
=> "false" 這個屬性，因此logstash默認把查詢結果的列明改成了小寫，同步進了es，因此就致使es裏面看到的字段名稱全是小寫。

最後總結：es是支持大寫字段名稱的，問題出在logstash沒用好，須要在同步配置中加上 lowercase_column_names => "false" 。記錄下來但願能夠幫到更多人。

3.2 同步到ES中的數據會不會重複？

想將關係數據庫的數據同步至ES中，若是在集羣的多臺服務器上同時啓動logstash。

解讀：實際項目中就是沒用隨機id 使用指定id做爲es的_id ，指定id能夠是url的md5.這樣相同數據就會走更新覆蓋之前數據

3.3 相同配置logstash,升級6.3以後不能同步數據。

解讀：高版本基於時間增量有優化。

tracking_column_type => "timestamp"應該是須要指定標識爲時間類型，默認爲數字類型numeric

3.4 ETL字段統一在哪處理?

解讀：能夠logstash同步mysql的時候sql查詢階段處理，如：select a_value as avalue***。

或者filter階段處理,mutate rename處理。

mutate {
        rename => ["shortHostname", "hostname" ]
    }

或者kafka階段藉助kafka stream處理。

四、小結

• 相關配置和同步都不復雜，複雜點每每在於filter階段的解析還有logstash性能問題。
• 須要結合實際業務場景作深刻的研究和性能分析。
• 有問題，歡迎留言討論。

推薦閱讀：
一、實戰 | canal 實現Mysql到Elasticsearch實時增量同步
二、乾貨 | Debezium實現Mysql到Elasticsearch高效實時同步
三、一張圖理清楚關係型數據庫與Elasticsearch同步 http://t.cn/EaAceD3
四、新的實現：http://t.cn/EaAt60O
五、mysql2mysql: http://t.cn/EaAtK7r
六、推薦開源實現：http://t.cn/EaAtjqN
加入星球，更短期更快習得更多幹貨！