ElasticsearchLogstash实现mysql同步数据到elasticsearch

2019独角兽企业重金招聘Python工程师标准>>>

有的时候&＃xff0c;我们在做查询时&＃xff0c;由于查询条件的多样、变化多端&＃xff08;比如根据时间查、根据名称模糊查、根据id查等等&＃xff09;&＃xff0c;或者查询的数据来自很多不同的库表或者系统&＃xff0c;这时就很难以一个较快的速度&＃xff08;几百毫秒&＃xff09;去从关系型数据库中直接获取我们想要的数据。

针对上面的情况&＃xff0c;可以考虑使用elasticsearch来进行数据的汇总&＃xff0c;然后提供给后台进行搜索&＃xff0c;可以大大提高检索的效率。

数据在存储在关系型数据库&＃xff08;如mysql&＃xff09;中&＃xff0c;我们怎样将这部分数据转移到elasticsearch中。这篇文章将介绍一个同步神器&＃xff1a;logstash-input-jdbc

安装

在官网下载最新的安装包&＃xff1a;
wget https://artifacts.elastic.co/downloads/logstash/logstash-6.2.4.tar.gz

解压并转移目录&＃xff1a;

tar zxvf logstash-6.2.4.tar.gz mv ./logstash-6.2.4 /usr/local/logstash

配置

安装插件
由于这里是从mysql同步数据到elasticsearch&＃xff0c;所以需要安装jdbc的入插件和elasticsearch的出插件&＃xff1a;logstash-input-jdbc、logstash-output-elasticsearch
安装效果图如下所示&＃xff1a;

下载mysql连接库
由于logstash是ruby开发的&＃xff0c;所以这里要下载mysql的连接库jar包&＃xff0c;从官网下载&＃xff0c;我这里下载的是&＃xff1a;mysql-connector-java-5.1.46.jar
将下载好的mysql-connector-java-5.1.46.jar&＃xff0c;放至/usr/local/logstash/config/目录下。
修改配置文件
在config目录下&＃xff0c;创建配置文件&＃xff08;logstash-mysql-es.conf&＃xff09;&＃xff1a;
input {jdbc {# mysql相关jdbc配置jdbc_connection_string &＃61;> "jdbc:mysql://10.112.76.30:3306/jack_test?useUnicode&＃61;true&characterEncoding&＃61;utf-8&useSSL&＃61;false"jdbc_user &＃61;> "root"jdbc_password &＃61;> "123456"# jdbc连接mysql驱动的文件目录&＃xff0c;可去官网下载:https://dev.mysql.com/downloads/connector/j/jdbc_driver_library &＃61;> "./config/mysql-connector-java-5.1.46.jar"# the name of the driver class for mysqljdbc_driver_class &＃61;> "com.mysql.jdbc.Driver"jdbc_paging_enabled &＃61;> truejdbc_page_size &＃61;> "50000"jdbc_default_timezone &＃61;>"Asia/Shanghai"# mysql文件, 也可以直接写SQL语句在此处&＃xff0c;如下&＃xff1a;# statement &＃61;> "select * from t_order where update_time >&＃61; :sql_last_value;"statement_filepath &＃61;> "./config/jdbc.sql"# 这里类似crontab,可以定制定时操作&＃xff0c;比如每分钟执行一次同步(分时天月年)schedule &＃61;> "* * * * *"#type &＃61;> "jdbc"# 是否记录上次执行结果, 如果为真,将会把上次执行到的 tracking_column 字段的值记录下来,保存到 last_run_metadata_path 指定的文件中#record_last_run &＃61;> true# 是否需要记录某个column 的值,如果record_last_run为真,可以自定义我们需要 track 的 column 名称&＃xff0c;此时该参数就要为 true. 否则默认 track 的是 timestamp 的值.use_column_value &＃61;> true# 如果 use_column_value 为真,需配置此参数. track 的数据库 column 名,该 column 必须是递增的. 一般是mysql主键tracking_column &＃61;> "update_time"tracking_column_type &＃61;> "timestamp"last_run_metadata_path &＃61;> "./logstash_capital_bill_last_id"# 是否清除 last_run_metadata_path 的记录,如果为真那么每次都相当于从头开始查询所有的数据库记录clean_run &＃61;> false#是否将字段(column) 名称转小写lowercase_column_names &＃61;> false} }output {elasticsearch {hosts &＃61;> "10.112.76.31:9200"index &＃61;> "mysql_order"document_id &＃61;> "%{id}"template_overwrite &＃61;> true}# 这里输出调试&＃xff0c;正式运行时可以注释掉stdout {codec &＃61;> json_lines} } 这里有几个注意点&＃xff1a;
&＃xff08;1&＃xff09;jdbc_driver_library
  mysql-connector-java-5.1.46.jar的存放目录&＃xff0c;这个一定要配置正确&＃xff0c;支持全路径和相对路径。如果配置不对&＃xff0c;将会报“can ”错误。
&＃xff08;2&＃xff09;sql_last_value
标志目前logstash同步的位置信息&＃xff08;类似offset&＃xff09;。比如id、updatetime。logstash通过这个标志&＃xff0c;可以判断目前同步到哪一条数据。
&＃xff08;3&＃xff09;statement、statement_filepath
  statement&＃xff1a;执行同步的sql语句&＃xff0c;可以同步部分数据。
   statement_filepath&＃xff1a;存储执行同步的sql语句。不和statement同时使用。
&＃xff08;4&＃xff09;schedule
定时器&＃xff0c;表示每隔多长时间同步一次数据。格式类似crontab。
&＃xff08;5&＃xff09;tracking_column、tracking_column_type
  tracking_column&＃xff1a;表示表中哪一列用于判断logstash同步的位置信息。与sql_last_value比较判断是否需要同步这条数据。
   tracking_column_type&＃xff1a;racking_column指定列的类型。支持两种类型&＃xff1a;numeric&＃xff08;默认&＃xff09;、timestamp。注意&＃xff1a;如果列是时间字段&＃xff08;比如updateTime&＃xff09;&＃xff0c;一定要指定这个类型为timestamp。我就踩了这个大坑。。。一直同步不成功&＃xff01;&＃xff01;&＃xff01;
&＃xff08;6&＃xff09;last_run_metadata_path
   存储sql_last_value值的文件名称及位置。
&＃xff08;7&＃xff09;document_id
生成elasticsearch的文档值&＃xff0c;尽量使用同步的数据中已有的唯一标识。比如同步订单数据&＃xff0c;可以使用订单号。