Elasticsearch增量同步Mysql数据

电玩女神 2021-11-23 08:54 453阅读 0赞

首先安装Elasticsearch，参考[https://blog.csdn.net/letterss/article/details/91361205][https_blog.csdn.net_letterss_article_details_91361205]

同步我们需要用到logstash工具，[下载logstash][logstash]

将下载的logstash-7.2.0.tar解压

**tar -zxvf logstash-7.2.0.tar**

将mysql的连接库jar包放到logstash-7.2.0/config目录下,我这里用的是mysql-connector-java-5.1.22-bin.jar

在logstash-7.2.0目录下面创建mysql.conf文件

**cd logstash-7.2.0**

**vim mysql.conf**

编写增量同步配置文件

\-----------------------------------------------------------------------------------------------------------------------------------------------------------

input \{  
  jdbc \{  
  jdbc\_driver\_library => "../config/mysql-connector-java-5.1.22-bin.jar"  
  jdbc\_driver\_class => "com.mysql.jdbc.Driver"  
  jdbc\_connection\_string => "jdbc:mysql://localhost:3306/itv\_basic\_v077"  
  jdbc\_user => "root"      
  jdbc\_password => "admin"

record\_last\_run => true  
  use\_column\_value => true  
  tracking\_column => "semi\_id" \#主键ID  
  last\_run\_metadata\_path => "/ES/log/dtv\_semi" \#存储记录位置  
  clean\_run => "false"

schedule => "\* \* \* \* \*"   \#定时字段 各字段含义（由左至右）分、时、天、月、年，全部为\*默认含义为每分钟都更新  
  statement => "select \* from dtv\_semi where semi\_id>:sql\_last\_value" \#sql语句,注意有冒号 这里的id就是上面的id  
  type => "con\_vod"  
  \}

jdbc \{  
  jdbc\_driver\_library => "./config/mysql-connector-java-5.1.22-bin.jar"  
  jdbc\_driver\_class => "com.mysql.jdbc.Driver"  
  jdbc\_connection\_string => "jdbc:mysql://localhost:3306/itv\_basic\_v077"  
  jdbc\_user => "root"      
  jdbc\_password => "admin"

record\_last\_run => true  
  use\_column\_value => true  
  tracking\_column => "id" \#主键ID  
  last\_run\_metadata\_path => "/ES/log/dtv\_semi\_con"  
  clean\_run => "false"

schedule => "\* \* \* \* \*"   \#定时字段 各字段含义（由左至右）分、时、天、月、年，全部为\*默认含义为每分钟都更新  
  statement => "select \* from dtv\_semi\_con where id>:sql\_last\_value" \#sql语句,注意有冒号 这里的id就是上面的id  
  type => "semi\_con"  
  \}

output \{

stdout \{  
        codec => json\_lines  
    \}  
   elasticsearch \{  
        index => "house" \#\#索引名称  
        document\_type => "%\{type\}"   \#这里是动态的引用上面配置的type  
        hosts => "localhost:9200"  \#Elasticsearch访问地址  
         document\_id => "%\{id\}"  \#主键字段  
    \}  
\}

\----------------------------------------------------------------------------------------------------------------------------------------------

然后我们在logstash-7.2.0目录启动

**bin/logstash -f mysql.conf **

如遇错误

Sending Logstash logs to /usr/local/logstash/logstash-6.5.0/logs which is now configured via log4j2.properties
    [2018-11-20T12:23:45,931][WARN ][logstash.config.source.multilocal] Ignoring the 'pipelines.yml' file because modules or command line options are specified
    [2018-11-20T12:23:46,088][FATAL][logstash.runner          ] Logstash could not be started because there is already another instance using the configured data directory.  If you wish to run multiple instances, you must change the "path.data" setting.
    [2018-11-20T12:23:46,130][ERROR][org.logstash.Logstash    ] java.lang.IllegalStateException: Logstash stopped processing because of an error: (SystemExit) exit

在data目录下面删除.lock文件

rm -rf data/.lock

然后重新启动

[https_blog.csdn.net_letterss_article_details_91361205]: https://blog.csdn.net/letterss/article/details/91361205
[logstash]: https://www.elastic.co/cn/products/logstash