0、題記

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

而mysql寫入kafka的選型方案有：

方案一：logstash_output_kafka 插件。

方案二：kafka_connector。

方案三：debezium 插件。

方案四：flume。

方案五：其他類似方案。

其中：debezium和flume是基于 mysql binlog 實現的。

如果需要同步歷史全量數據+實時更新數據，建議使用logstash。

1、logstash同步原理

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

實際上， 核心logstash的同步原理的掌握 ，有助于大家理解類似的各種庫之間的同步。

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

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

input?{?}

filter?{?}

output?{?}

1.1 input輸入

包含但遠不限于：

jdbc：關系型數據庫：mysql、 oracle 等。

file：從文件系統上的文件讀取。

syslog：在已知端口514上偵聽syslog消息。

redis：redis消息。beats：處理 Beats發送的事件。

kafka：kafka實時數據流。

1.2 filter過濾器

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

可以把它比作數據處理的 ETL 環節。

一些有用的過濾包括：

grok：解析并構造任意文本。 Grok是目前Logstash中將非結構化日志數據解析為結構化和可查詢內容的最佳方式 。有了內置于Logstash的120種模式，您很可能會找到滿足您需求的模式！

mutate：對事件字段執行常規轉換。您可以重命名，刪除，替換和修改事件中的字段。

drop：完全刪除事件，例如調試事件。

clone：制作事件的副本，可能添加或刪除字段。

geoip：添加有關IP地址的地理位置的信息。

1.3 output輸出

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

elasticsearch：將事件數據發送到Elasticsearch。

file：將事件數據寫入磁盤上的文件。

kafka：將事件寫入Kafka。

詳細的filter demo參考：http://t.cn/EaAt4zP

2、同步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、坑總結

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處理。

4、小結

相關配置和同步都不復雜，復雜點往往在于filter階段的解析還有logstash性能問題。

需要結合實際業務場景做深入的研究和性能分析。

有問題，歡迎留言討論。

推薦閱讀：

3、 一張圖理清楚關系型數據庫與Elasticsearch同步?http://t.cn/EaAceD3

4、新的實現：http://t.cn/EaAt60O

5、mysql2mysql:?http://t.cn/EaAtK7r 6、推薦開源實現： http://t.cn/EaAtjqN

加入星球，更短時間更快習得更多干貨！

創作挑戰賽新人創作獎勵來咯，堅持創作打卡瓜分現金大獎

總結

以上是生活随笔為你收集整理的日志文件和mysql同步到kafka_logstash_output_kafka:Mysql 同步 Kafka 深入详解的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。