3、安装Logstash的plugin，mysql数据同步

插件安装、卸载、更新的命令。

官网&＃xff1a;https://www.elastic.co/guide/en/logstash/current/working-with-plugins.html#installing-local-plugins

1、查询已安装的插件&＃xff1a;

Logstash发行包捆绑常见的插件&＃xff0c;以便您可以开箱即用。可以查询当前可用的插件&＃xff1a;

bin/logstash-plugin list 将列出所有安装的插件
bin/logstash-plugin list --verbose 将列出已安装的插件与版本信息
bin/logstash-plugin list &＃39;*namefragment*&＃39; 将列出所有安装的包含namefragment的插件
bin/logstash-plugin list --group output 将列出特定组的所有安装的插件&＃xff08;输入&＃xff0c;过滤器&＃xff0c;编解码器&＃xff0c;输出&＃xff09;

2、安装插件&＃xff1a;

线上安装插件:

bin/logstash-plugin install logstash-output-kafka

线下安装插件&＃xff1a;

bin/logstash-plugin install /path/to/logstash-output-kafka-1.0.0.gem

3、更新插件&＃xff1a;

bin/logstash-plugin update 更新所有已经安装的插件
bin/logstash-plugin update logstash-output-kafka 更新插件logstash-output-kafka

4、移除插件&＃xff1a;

bin/logstash-plugin remove logstash-output-kafka 二、实现mysql数据同步。

1、安装相关插件&＃xff1a;

input我这里使用logstash-input-jdbc插件&＃xff0c;output使用logstash-output-jdbc插件。因为logstash自带了logstash-input-jdbc插件。所以这里只安装logstash-output-jdbc。

我这里使用线上安装&＃xff1a;

bin/logstash-plugin install logstash-output-jdbc

1、下载插件。下载地址&＃xff1a;https://github.com/theangryangel/logstash-output-jdbc。

2、线下安装&＃xff0c;参考下面2个博客&＃xff1a;

        http://blog.csdn.net/yeyuma/article/details/50240595

        https://my.oschina.net/MrYx3en/blog/508230

2、配置文件&＃xff1a;

input {stdin { } jdbc {jdbc_driver_library &＃61;> "G:/MvnRepository/mysql/mysql-connector-java/5.1.41/mysql-connector-java-5.1.41.jar"jdbc_driver_class &＃61;> "com.mysql.jdbc.Driver"jdbc_connection_string &＃61;> "jdbc:mysql://localhost:3306/xnc"jdbc_user &＃61;> "root"jdbc_password &＃61;> "123456"# or jdbc_password_filepath &＃61;> "/path/to/my/password_file"# where p.update_time >&＃61; :sql_last_startstatement &＃61;> "SELECT id, product_spec_id,zone_id,recorded_by,CAST(price_per_unit AS CHAR) price_per_unit,uom,latest,recording_date,create_time,update_time from price p"jdbc_paging_enabled &＃61;> "true"jdbc_page_size &＃61;> "50000"}
}
output {#标准输出。为了测试stdout { codec &＃61;> rubydebug }jdbc {driver_class &＃61;> "com.mysql.jdbc.Driver"driver_jar_path &＃61;> "G:/MvnRepository/mysql/mysql-connector-java/5.1.41/mysql-connector-java-5.1.41.jar"connection_string &＃61;> "jdbc:mysql://localhost:3306/xncprice"username &＃61;> "root"password &＃61;> "123456"statement &＃61;> [ "INSERT INTO price (id, product_spec_id, zone_id,recorded_by,price_per_unit,uom,latest,recording_date,create_time,update_time) VALUES(?,?,?,?,CAST(? AS decimal),?,?,?,?,?)", "id", "product_spec_id", "zone_id", "recorded_by","price_per_unit", "uom" , "latest" , "recording_date" , "create_time" , "update_time" ]}}

3、运行。

logstash -f G:\logstash-5.5.1\config\my-logstash.conf

这里需要注意数据库类型。因为我的price_per_unit是decimal类型的&＃xff0c;在运行之后&＃xff0c;price_per_unit插入的数据一直是null&＃xff0c;改了目标类型为double还是不行。解决方式有2个&＃xff1a;

1. 改源类型&＃xff08;input中price_per_unit的数据类型&＃xff09;为double。
2. 查询的时候把源的decimal转换为char类型&＃xff0c;插入目标的时候再转为decimal。&＃xff08;我使用这个&＃xff09;。

4、同步update操作。

把output使用REPLACE 语句&＃xff0c;如果数据id重复时候&＃xff0c;直接用后面的数据替换前面的数据。

output {#标准输出。为了测试stdout { codec &＃61;> rubydebug }jdbc {driver_class &＃61;> "com.mysql.jdbc.Driver"driver_jar_path &＃61;> "G:/MvnRepository/mysql/mysql-connector-java/5.1.41/mysql-connector-java-5.1.41.jar"connection_string &＃61;> "jdbc:mysql://localhost:3306/xncprice"username &＃61;> "root"password &＃61;> "123456"statement &＃61;> [ "REPLACE INTO price (id, product_spec_id, zone_id,recorded_by,price_per_unit,uom,latest,recording_date,create_time,update_time) VALUES(?,?,?,?,CAST(? AS decimal),?,?,?,?,?)", "id", "product_spec_id", "zone_id", "recorded_by","price_per_unit", "uom" , "latest" , "recording_date" , "create_time" , "update_time" ]}