logstash mysql 同步到 elasticsearch

Bertha 。 2022-03-02 15:59 338阅读 0赞

es，kinana环境参考：[https://blog.csdn.net/qq\_38270106/article/details/88394869][https_blog.csdn.net_qq_38270106_article_details_88394869]

es集群环境参考：[https://blog.csdn.net/qq\_38270106/article/details/88411221][https_blog.csdn.net_qq_38270106_article_details_88411221]

Logstash是一个开源数据收集引擎，具有实时管道功能。Logstash可以动态地将来自不同数据源的数据统一起来，并将数据标准化到你所选择的目的地

1.上传logstash-6.4.3.tar.gz到服务中

2.tar –zxvf  logstash-6.4.3.tar.gz

3.cd logstash-6.4.3

4. bin/logstash-plugin install logstash-input-jdbc

5. bin/logstash-plugin install logstash-output-elasticsearch

6. vi mysql.conf

> input \{  
>   jdbc \{  
>     jdbc\_driver\_library => "/usr/local/sql/mysql-connector-java-5.1.46.jar"  
>     jdbc\_driver\_class => "com.mysql.jdbc.Driver"  
>     jdbc\_connection\_string => "jdbc:mysql://192.168.1.103:3306/test\_elk"  
>     jdbc\_user => "root"  
>     jdbc\_password => "root"  
>     schedule => "\* \* \* \* \*"  
>     statement => "SELECT \* FROM user WHERE update\_time >= :sql\_last\_value"  
>     use\_column\_value => true  
>     tracking\_column\_type => "timestamp"  
>     tracking\_column => "update\_time"  
>     last\_run\_metadata\_path => "syncpoint\_table"  
>   \}  
> \}
> 
>   
> output \{  
>     elasticsearch \{  
>         \# ES的IP地址及端口  
>         hosts => \["192.168.128.130:9200", "192.168.128.130:9201"\]  
>         \# 索引名称 可自定义  
>         index => "user"  
>         \# 需要关联的数据库中有有一个id字段，对应类型中的id  
>         document\_id => "%\{id\}"  
>         document\_type => "user"  
>     \}  
>     stdout \{  
>         \# JSON格式输出  
>         codec => json\_lines  
>     \}  
> \}

说明：

jdbc\_driver\_library: jdbc mysql 驱动的路径，在上一步中已经下载  
jdbc\_driver\_class: 驱动类的名字，mysql 填 com.mysql.jdbc.Driver 就好了  
jdbc\_connection\_string: mysql 地址  
jdbc\_user: mysql 用户  
jdbc\_password: mysql 密码  
schedule: 执行 sql 时机，类似 crontab 的调度  
statement: 要执行的 sql，以 “:” 开头是定义的变量，可以通过 parameters 来设置变量，这里的 sql\_last\_value 是内置的变量，表示上一次 sql 执行中 update\_time 的值，这里 update\_time 条件是 >= 因为时间有可能相等，没有等号可能会漏掉一些增量  
use\_column\_value: 使用递增列的值  
tracking\_column\_type: 递增字段的类型，numeric 表示数值类型, timestamp 表示时间戳类型  
tracking\_column: 递增字段的名称，这里使用 update\_time 这一列，这列的类型是 timestamp  
last\_run\_metadata\_path: 同步点文件，这个文件记录了上次的同步点，重启时会读取这个文件，这个文件可以手动修改

7. 创建user表

> CREATE TABLE \`user\` (  
>   \`id\` bigint(20) NOT NULL AUTO\_INCREMENT,  
>   \`name\` varchar(255) NOT NULL,  
>   \`update\_time\` timestamp NULL DEFAULT NULL ON UPDATE CURRENT\_TIMESTAMP,  
>   PRIMARY KEY (\`id\`)  
> ) ENGINE=InnoDB DEFAULT CHARSET=utf8;

8.上传mysql jar到/usr/local/sql/ mysql-connector-java-5.1.46.jar

9. 启动logstash

> ./bin/logstash -f mysql.conf

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4MjcwMTA2_size_16_color_FFFFFF_t_70][] 数据库插入两条记录：

![2019032021481990.png][]

同步log打印：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4MjcwMTA2_size_16_color_FFFFFF_t_70 1][]

kinana查看es的数据：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4MjcwMTA2_size_16_color_FFFFFF_t_70 2][]

## 多文件方式同步ES数据 ##

一个 logstash 实例可以借助 pipelines 机制同步多个表，只需要写多个配置文件就可以了，假设我们有两个表 table1 和 table2，对应两个配置文件 mysql\_user.conf 和 mysql\_member.conf，

1. 创建member表

> CREATE TABLE \`member\` (  
>   \`id\` bigint(20) NOT NULL AUTO\_INCREMENT,  
>   \`name\` varchar(255) NOT NULL,  
>   \`update\_time\` timestamp NULL DEFAULT NULL ON UPDATE CURRENT\_TIMESTAMP,  
>   PRIMARY KEY (\`id\`)  
> ) ENGINE=InnoDB DEFAULT CHARSET=utf8;

2.上传到 /usr/local/sql目录下

mysql\_user.conf

mysql\_member.conf

> input \{  
>   jdbc \{  
>     jdbc\_driver\_library => "/usr/local/sql/mysql-connector-java-5.1.46.jar"  
>     jdbc\_driver\_class => "com.mysql.jdbc.Driver"  
>     jdbc\_connection\_string => "jdbc:mysql://192.168.1.103:3306/test\_elk"  
>     jdbc\_user => "root"  
>     jdbc\_password => "root"  
>     schedule => "\* \* \* \* \*"  
>     statement => "SELECT \* FROM member WHERE update\_time >= :sql\_last\_value"  
>     use\_column\_value => true  
>     tracking\_column\_type => "timestamp"  
>     tracking\_column => "update\_time"  
>     last\_run\_metadata\_path => "syncpoint\_table"  
>   \}  
> \}
> 
>   
> output \{  
>     elasticsearch \{  
>         \# ES的IP地址及端口  
>         hosts => \["192.168.128.130:9200", "192.168.128.130:9201"\]  
>         \# 索引名称 可自定义  
>         index => "member"  
>         \# 需要关联的数据库中有有一个id字段，对应类型中的id  
>         document\_id => "%\{id\}"  
>         document\_type => "member"  
>     \}  
>     stdout \{  
>         \# JSON格式输出  
>         codec => json\_lines  
>     \}  
> \}

只需要修改sql语句，es索引和文档类型

2.在 config/pipelines.yml 中配置

> cd conf
> 
> vi pipelines.yml

\- pipeline.id: table1  
  path.config: "/usr/local/sql/mysql\_user.conf"  
\- pipeline.id: table2  
  path.config: "/usr/local/sql/mysql\_member.conf"

![20190320220215240.png][]

3.启动

> cd /usr/local/logstash-6.4.3
> 
> ./bin/logstash

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4MjcwMTA2_size_16_color_FFFFFF_t_70 3][] member表插入两条数据：

![20190320223140147.png][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4MjcwMTA2_size_16_color_FFFFFF_t_70 4][]

kinana查看es的同步数据![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4MjcwMTA2_size_16_color_FFFFFF_t_70 5][]

[https_blog.csdn.net_qq_38270106_article_details_88394869]: https://blog.csdn.net/qq_38270106/article/details/88394869
[https_blog.csdn.net_qq_38270106_article_details_88411221]: https://blog.csdn.net/qq_38270106/article/details/88411221
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4MjcwMTA2_size_16_color_FFFFFF_t_70]: /images/20220302/af772922bb234877a6dfe778c2b70bb8.png
[2019032021481990.png]: /images/20220302/a76e84879fad440cb93dfc4ef15797f7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4MjcwMTA2_size_16_color_FFFFFF_t_70 1]: /images/20220302/eaac7d8c17f74731bc4fcfbe16b3fa99.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4MjcwMTA2_size_16_color_FFFFFF_t_70 2]: /images/20220302/e38e26b44990438db7bc1a43bcc89415.png
[20190320220215240.png]: /images/20220302/1cc279cd6b5c4b59b9b77acb9c8ea860.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4MjcwMTA2_size_16_color_FFFFFF_t_70 3]: /images/20220302/4fff0ff536cd42fd8d09424fdc7ba432.png
[20190320223140147.png]: /images/20220302/59e477da0b5044488dc57fd450b4fb25.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4MjcwMTA2_size_16_color_FFFFFF_t_70 4]: /images/20220302/c4abccc185d2460698d6d22a0341f1ef.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4MjcwMTA2_size_16_color_FFFFFF_t_70 5]: /images/20220302/6b2338e9eda24bf282524dcabff7c507.png