Elasticsearch - Logstash实现mysql同步数据到elasticsearch

你的名字 2023-06-10 03:21 3阅读 0赞

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

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

数据在存储在关系型数据库（如mysql）中，我们怎样将这部分数据转移到elasticsearch中。这篇文章将介绍一个同步神器：**logstash-input-jdbc**

> **安装**
> 
> 1.  在[官网][Link 1]下载最新的安装包：
>     
>         wget https://artifacts.elastic.co/downloads/logstash/logstash-6.2.4.tar.gz
> 2.  解压并转移目录：
>     
>         tar zxvf logstash-6.2.4.tar.gz
>         mv ./logstash-6.2.4 /usr/local/logstash
>     
>

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

> **启动**
> 
> 在根目录下，执行命令：
> 
>     nohup bin/logstash -f config/logstash-mysql-es.conf > logs/logstash.out &
> 
> 效果图如下：
> 
> ![format_png 2][]

> **同步**
> 
> ![format_png 3][]
> 
> 完成了一条数据的同步

[Link 1]: https://www.elastic.co/cn/products/logstash
[format_png]: /images/20230531/8fa8e9cc66194c7f8cd6a4f34c19d358.png
[format_png 1]: /images/20230531/95255097f1624eea8ac4d21a94d1540a.png
[Link 2]: https://dev.mysql.com/downloads/connector/j/
[jdbc_driver_library]: https://www.elastic.co/guide/en/logstash/6.2/plugins-inputs-jdbc.html#plugins-inputs-jdbc-jdbc_driver_library
[statement]: https://www.elastic.co/guide/en/logstash/6.2/plugins-inputs-jdbc.html#plugins-inputs-jdbc-statement
[statement_filepath]: https://www.elastic.co/guide/en/logstash/6.2/plugins-inputs-jdbc.html#plugins-inputs-jdbc-statement_filepath
[schedule]: https://www.elastic.co/guide/en/logstash/6.2/plugins-inputs-jdbc.html#plugins-inputs-jdbc-schedule
[tracking_column]: https://www.elastic.co/guide/en/logstash/6.2/plugins-inputs-jdbc.html#plugins-inputs-jdbc-tracking_column
[tracking_column_type]: https://www.elastic.co/guide/en/logstash/6.2/plugins-inputs-jdbc.html#plugins-inputs-jdbc-tracking_column_type
[last_run_metadata_path]: https://www.elastic.co/guide/en/logstash/6.2/plugins-inputs-jdbc.html#plugins-inputs-jdbc-last_run_metadata_path
[document_id]: https://www.elastic.co/guide/en/logstash/6.2/plugins-outputs-elasticsearch.html#plugins-outputs-elasticsearch-document_id
[format_png 2]: /images/20230531/32704a6e810d4e97b51ca1ab99ea1f3d.png
[format_png 3]: /images/20230531/6cb173e13c8e48a99f84e4aef88d0613.png