mysql2es全量更新方案

Posted 2021-03-03 呢喃的歌声

mysql到es的全量更新方案可以通过 logstash来实现

logstash可以做基于全量的定时更新，也可以做基于时间的定时更新。

logstash的使用方式如下（本人用的是logstash-7.6.1,不同版本在使用上应该有细微区别,区别百度可解决）

1.https://artifacts.elastic.co/downloads/logstash/logstash-7.6.1.zip 下载安装包

2.上传到linux 通过unzip 解压到bin目录下新建mysql目录（存放配置文件，由于ES7.6.1默认不让安装root,我也一起放到新用户es下，目录权限也全部赋值了）

3.安装logstash的JDBC插件（bin/logstash-plugin install logstash-integration-jdbc）

4.在mysql目录中放入jdbc.conf（连接数据库的配置）  jdbc.sql（编写sql的配置,通过此语句查询到mysql数据）  last_value_meta（基于时间点更新需要用到）  mysql-connector-java-5.1.35.jar（JDBC包）

三个文件的内容贴这里以供参考

jdbc.conf：

input {
jdbc {
#set timezone
jdbc_default_timezone => "Asia/Shanghai"
# mysql 数据库链接,dianpingdb为数据库名
jdbc_connection_string => "jdbc:mysql://192.168.1.4:3306/dianping"
# 用户名和密码
jdbc_user => "root"
jdbc_password => "root"
# 驱动
jdbc_driver_library => "/opt/logstash-7.6.1/bin/mysql/mysql-connector-java-5.1.35.j
ar"
# 驱动类名
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_paging_enabled => "true"
jdbc_page_size => "50000"
last_run_metadata_path => "/opt/logstash-7.6.1/bin/mysql/last_value_meta"
# 执行的sql 文件路径+名称
statement_filepath => "/opt/logstash-7.6.1/bin/mysql/jdbc.sql"
# 设置监听间隔 各字段含义（由左至右）分、时、天、月、年，全部为*默认含义为每分钟都
更新
schedule => "* * * * *"
}
}

 

  jdbc.sql：

select a.id,a.name,a.tags,concat(a.latitude,‘,‘,a.longitude) as location,a.remark_score,a.price_per_man,a.category_id,b.name as category_name,a.seller_id,c.remark_score as seller_remark_score,c.disable_flag as seller_disabled_flag from shop a inner join category b on a.category_id = b.id inner join seller c on c
.id = a.seller_id where a.updated_at > :sql_last_value or b.updated_at > :sql_last_value or c.updated_at > :sql_last_value

last_value_meta（此文件一开始只填入了时间）：

--- !ruby/object:DateTime ‘2020-03-29 08:13:00.216387000 Z‘   

 

以上是基于最新时间点,用轮训方式进行更新mysql库数据到es库中

 

 

启动方式到bin下./logstash -f mysql/jdbc.conf