Elasticsearch使用Logstash-input-jdbc同步mysql数据（全量和增量）

超、凢脫俗 2022-04-05 17:58 443阅读 0赞

--------------------

作者：camelcanoe  
来源：CSDN  
原文：[https://blog.csdn.net/camelcanoe/article/details/79759376][https_blog.csdn.net_camelcanoe_article_details_79759376]  
版权声明：本文为博主原创文章，转载请附上博文链接！

项目中用到elasticsearch，初始化数据时时写的程序从数据库里面查询出来，然后多线程往elasticsearch里面写入的。

今天试了一下Logstash-input-jdbc插件，发现高效又方便，而且可以设置定时任务。

1、安装插件

在logstash的bin目录下执行命令： logstash-plugin install logstash-input-jdbc

2、配置文件和jar包  
在bin目录下新建一个config-mysql目录,里面包含mysql.conf,  
在H:\\software\\logstash-6.1.2\\logstash-6.1.2\\lib 加入mysql的驱动 mysql-connector-java-5.1.38.jar

mysql.conf的内容如下：

input \{  
stdin \{  
\}  
jdbc \{  
\# 数据库  
jdbc\_connection\_string => “jdbc:mysql://39.107.60.74:3306/db\_plat3”  
\# 用户名密码  
jdbc\_user => “root”  
jdbc\_password => “letsgo123QWE”  
\# jar包的位置  
jdbc\_driver\_library => “H:\\software\\logstash-6.1.2\\logstash-6.1.2\\lib\\mysql-connector-java-5.1.38.jar”  
\# mysql的Driver  
jdbc\_driver\_class => “com.mysql.jdbc.Driver”  
jdbc\_paging\_enabled => “true”  
jdbc\_page\_size => “50000”  
\#statement\_filepath => “config-mysql/test02.sql”  
statement => “select \* from information”  
schedule => “\* \* \* \* \*”  
\#索引的类型  
type => “information”  
\}  
\}

filter \{  
json \{  
source => “message”  
remove\_field => \[“message”\]  
\}  
\}

output \{  
elasticsearch \{  
hosts => “192.168.1.70:9200”  
\# index名  
index => “information”  
\# 需要关联的数据库中有有一个id字段，对应索引的id号  
document\_id => “%\{id\}”  
\}  
stdout \{  
codec => json\_lines  
\}  
\}

3、启动logstash

然后通过以下命令启动logstash

.\\logstash.bat -f .\\config-mysql\\mysql.conf

过一会他就会自动的往ES里添加数据。

4、增量索引

但是现在有一个问题是：往elasticsearch里面写入是全量的，需要改成增量。

修改mysql.conf的内容如下

#使用其它字段追踪，而不是用时间
      use_column_value => true
      #追踪的字段
      tracking_column => id
      record_last_run => true
     #上一个sql_last_value值的存放文件路径, 必须要在文件中指定字段的初始值

last\_run\_metadata\_path=>“H:\\software\\logstash-6.1.2\\logstash-6.1.2\\bin\\config-mysql\\station\_parameter.txt”  
\#开启分页查询  
jdbc\_paging\_enabled => “true”  
jdbc\_page\_size => “50000”  
statement\_filepath => “config-mysql/information.sql”  
\#statement => "select \* from information where id > :sql\_last\_value "  
schedule => “\* \* \* \* \*”  
\#索引的类型  
type => “information”  
\}  
\}

filter \{  
json \{  
source => “message”  
remove\_field => \[“message”\]  
\}  
\}

修改的内容见红色。

station\_parameter.txt中的内容如下：

logstash执行时打印出来的sql如下：

其中遇到的一个问题是用：

statement => "select \* from information where id > :sql\_last\_value "

时会报:sql\_last\_value 的错 ，暂时不知道怎么解决。于是改用文件config-mysql/information.sql存在sql语句。

ps :

上述问题是因为从人家那里copy过来的时候是错的 把 :last\_sql\_value 改成 :sql\_last\_value 就可以了。

5、用时间来实现增量：

input \{  
stdin \{  
\}  
jdbc \{

# 数据库 #

jdbc\_connection\_string => “jdbc:mysql://39.107.60.74:3306/db\_plat3”

# 用户名密码 #

jdbc\_user => “root”  
jdbc\_password => “letsgo123QWE”

# jar包的位置 #

jdbc\_driver\_library => “H:\\software\\logstash-6.1.2\\logstash-6.1.2\\lib\\mysql-connector-java-5.1.38.jar”

# mysql的Driver #

jdbc\_driver\_class => “com.mysql.jdbc.Driver”  
\#开启分页查询  
jdbc\_paging\_enabled => “true”  
jdbc\_page\_size => “50000”  
\#statement\_filepath => “config-mysql/information.sql”  
statement => “select \* from information where modify\_time > :sql\_last\_value”  
schedule => “\* \* \* \* \*”  
\#索引的类型  
type => “information”  
\}  
\}  
filter \{  
json \{  
source => “message”  
remove\_field => \[“message”\]  
\}  
\}  
output \{  
elasticsearch \{  
hosts => “192.168.1.70:9200”

# index名 #

index => “information”

# 需要关联的数据库中有有一个id字段，对应索引的id号 #

document\_id => “%\{id\}”  
\}  
stdout \{  
codec => json\_lines  
\}  
\}  
modify\_time 是我表里的最后修改时间字段。  
而 :sql\_last\_value如果input里面use\_column\_value => true， 即如果设置为true的话，可以是我们设定的字段的上一次的值。  
默认 use\_column\_value => false， 这样 :sql\_last\_value为上一次更新的最后时刻值。  
也就是说，对于新增的值，才会更新。这样就实现了增量更新的目的。

[https_blog.csdn.net_camelcanoe_article_details_79759376]: https://blog.csdn.net/camelcanoe/article/details/79759376