利用Logstash实现ES和MySQL同步

1. ES和MySQL同步方法

1. 编程式

   • 使用之前的elasticdump&＃xff0c;手动变成生成json文件&＃xff0c;编写shellscript导入ES。后续的数据新增使用PHP或kava客户端进行编程触发同步。

2. 使用插件&＃xff08;没有什么特别好的插件&＃xff09;

   • 国人编写的go-mysql-elasticsearch&＃xff08;go开发&＃xff09;
   • 官方推荐基于logstash的插件logstash-input-jdbc

3. 自己写

   • 采用阿里巴巴的一个开源项目canal&＃xff0c;可以模拟mysql从服务器。封装了bin-log分析&＃xff0c;我们可以自己开发一个客户端可以更好地控制数据同步。

2. logstash-input-jdbc同步方法

1. 下载mysql&＃43;java驱动包&＃xff1a;https://dev.mysql.com/downloads/connector/j/

2. 配置文件

   input {jdbc {jdbc_driver_library &＃61;> "mysql-connector-java-5.1.36-bin.jar"jdbc_driver_class &＃61;> "com.mysql.jdbc.Driver"jdbc_connection_string &＃61;> "jdbc:mysql://localhost:3306/mydb"jdbc_user &＃61;> "mysql"parameters &＃61;> { "favorite_artist" &＃61;> "Beethoven" }schedule &＃61;> "* * * * *"statement &＃61;> "SELECT * from songs where artist &＃61; :favorite_artist"}}


   • 文档&＃xff1a;https://www.elastic.co/guide/en/logstash/master/plugins-inputs-jdbc.html
   • 关于 schedule &＃61;> "* * * * *"
     • 类似于Linux的crond的配置
     • 从左往右 分 时 日 月 年 &＃xff08;最小单位&＃xff1a;分&＃xff09;
     • 全部写*表示每分钟都更新
     • • 5 * * * 表示每天5点钟更新
     • • 5 * 1 * 表示一月份每天5点更新

3. 准备数据&＃xff0c;创建mysql表jt_news

   CREATE TABLE &＃96;jt_news&＃96; (&＃96;news_id&＃96; int(11) NOT NULL AUTO_INCREMENT,&＃96;news_title&＃96; varchar(64) DEFAULT NULL,&＃96;news_abstract&＃96; varchar(64) DEFAULT NULL,&＃96;news_updatetime&＃96; datetime DEFAULT NULL,&＃96;news_clicknum&＃96; int(3) DEFAULT NULL,&＃96;news_class&＃96; varchar(64) DEFAULT NULL,PRIMARY KEY (&＃96;news_id&＃96;)) ENGINE&＃61;MyISAM AUTO_INCREMENT&＃61;4 DEFAULT CHARSET&＃61;utf8;


   INSERT INTO &＃96;jt_news&＃96; VALUES (1,&＃39;Java开发新闻&＃39;,&＃39;Java新闻摘要&＃39;,&＃39;2017-09-28 11:14:49&＃39;,11,&＃39;编程语言&＃39;),(2,&＃39;PHP开发新闻&＃39;,&＃39;PHP新闻摘要&＃39;,&＃39;2017-09-28 11:15:13&＃39;,13,&＃39;web开发&＃39;),(3,&＃39;js前后端分离&＃39;,&＃39;Javascript&＃39;,&＃39;2017-09-28 11:15:38&＃39;,1,&＃39;web开发&＃39;);


4. 创建索引&＃xff08;mapping&＃xff09;

   PUT jtthink-test{"mappings": {"jt_news": {"properties": {"news_title": {"type": "text","analyzer": "ik_max_word"},"news_abstract": {"type": "text","analyzer": "ik_max_word"},"news_updatetime": {"type": "date"},"news_clicknum": {"type": "integer"},"news_class": {"type": "keyword"}}}}}


5. 创建配置文件 mysql.conf

   • 在logstash/bin下创建&＃xff08;该目录可变&＃xff09;

   • 文件内容

     input {jdbc {jdbc_driver_library &＃61;> "/usr/local/drivers/mysql-connector-java-5.1.44/mysql-connector-java-5.1.44-bin.jar"jdbc_driver_class &＃61;> "com.mysql.jdbc.Driver"jdbc_connection_string &＃61;> "jdbc:mysql://localhost:3306/jtthink"jdbc_user &＃61;> "root"jdbc_password &＃61;> "shadowolf1995."statement &＃61;> "SELECT * from jt_news order by news_id desc"tracking_column &＃61;> "news_updatetime"tracking_column_type &＃61;> "timestamp"schedule &＃61;> "* * * * *"}}filter {}output {stdout {codec &＃61;> rubydebug}elasticsearch {hosts &＃61;> ["127.0.0.1:9201"]index &＃61;> "jtthink-test"document_id &＃61;> "%{news_id}"}}


6. 启动logstash

   ./logstash -f mysql.conf


7. 使用kibana创建index jtthink-test&＃xff0c;查看数据。