Logstash同步Mysql至ES
问题描述
程序推送商品信息至Elasticsearch,因为同步的问题，导致了商品信息不同步
解决方案
使用Logstash同步Mysql数据库数据
使用软件
- jdk8
- logstash 7.7.1
- Elasticsearch
- mysql-connector-java-8.0.13.jar
安装过程
jdk8的安装
# 安装jdk8，这个链接下不了，自己去oracle官网下jdk8的tar包
wget https://download.oracle.com/otn/java/jdk/8u333-b02/2dee051a5d0647d5be72a7c0abff270e/jdk-8u333-linux-x64.tar.gz?AuthParam=1653980946_5c1e711df3459862c6ed1accab4682b6
tar -xf jdk-8u333-linux-x64.tar.gz -C /usr/local/
mv /usr/local/jdk1.8.0_333 /usr/local/jdk
# 输出环境变量至文件内
cat >> /etc/profile << EOF
export JAVA_HOME=/usr/local/jdk
export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar:$JAVA_HOME/jre/lib
export PATH=$JAVA_HOME/bin:$JAVA_HOME/jre/bin:$PATH
EOF
source /etc/profile
# 测试java是否安装成功
java -version
java version "1.8.0_333"
Java(TM) SE Runtime Environment (build 1.8.0_333-b02)
Java HotSpot(TM) 64-Bit Server VM (build 25.333-b02, mixed mode)
logstash的安装
wget https://artifacts.elastic.co/downloads/logstash/logstash-7.7.1.tar.gz
tar -xf logstash-7.7.1.tar.gz -C /usr/local
mv /usr/local/logstash-7.7.1 /usr/local/logstash
# 安装java的jdbc
curl --create-dirs -sSLo /usr/local/logstash/java/mysql-connector-java-8.0.13.zip https://cdn.mysql.com//archives/mysql-connector-java-8.0/mysql-connector-java-8.0.13.zip
cd /usr/local/logstash/java/
unzip mysql-connector-java-8.0.13.zip
# 修改logstash.yml配置文件
# 新增2行内容
path.config: /usr/local/logstash/pipeline/*.conf #定义配置文件地址
path.logs: /var/log/logstash #定义日志文件
mkdir /usr/local/logstash/pipeline
/usr/local/logstash/pipeline/mall_goods_index.json 配置文件编写
{
"template":"mall_goods_index",
"order":1,
"settings":{
"number_of_shards":5,
"number_of_replicas":0,
"refresh_interval":"60s"
},
"mappings ":{
"normal":{
"dynamic":"strict",
"properties":{
"id":{
"type":"long"
},
"goodsCode":{
"type":"string"
},
"goodsId":{
"type":"long"
},
"shopId":{
"type":"long"
},
"goodsName":{
"type":"string"
},
"pinyin":{
"type":"keyword"
},
"goodsImgs":{
"type":"string"
},
"killPrice":{
"type":"double"
},
"price":{
"type":"double"
},
"spenBuild":{
"type":"string"
},
"defaultSpec":{
"type":"string"
},
"categoryIds":{
"type":"string"
},
"categoryNames":{
"type":"string"
},
"status":{
"type":"integer"
},
"stock":{
"type":"double"
},
"searchInfo":{
"type":"string"
},
"updateTime":{
"type":"keyword"
},
"orderCount":{
"type":"keyword"
}
}
}
}
}
/usr/local/logstash/pipeline/test_goods_index.json 编辑es格式
{
"template":"test_goods_index",
"order":1,
"settings":{
"number_of_shards":5,
"number_of_replicas":0,
"refresh_interval":"60s"
},
"mappings ":{
"normal":{
"dynamic":"strict",
"properties":{
"id":{
"type":"long"
},
"goodsCode":{
"type":"string"
},
"goodsId":{
"type":"long"
},
"shopId":{
"type":"long"
},
"goodsName":{
"type":"string"
},
"pinyin":{
"type":"keyword"
},
"goodsImgs":{
"type":"string"
},
"killPrice":{
"type":"double"
},
"price":{
"type":"double"
},
"spenBuild":{
"type":"string"
},
"defaultSpec":{
"type":"string"
},
"categoryIds":{
"type":"string"
},
"categoryNames":{
"type":"string"
},
"status":{
"type":"integer"
},
"stock":{
"type":"double"
},
"searchInfo":{
"type":"string"
},
"updateTime":{
"type":"keyword"
},
"orderCount":{
"type":"keyword"
}
}
}
}
}
/usr/local/logstash/pipeline/mysql-to-many-es.conf
input{
jdbc {
type => "cm1" #定义客户1的数据源
jdbc_driver_library => "/usr/local/logstash/java/mysql-connector-java-8.0.13.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_connection_string => "xxxx" #修改数据库地址
jdbc_user => "xxxx"
jdbc_password => "xxxxx"
# 开启分页
jdbc_paging_enabled => true
tracking_column => "unix_ts_in_secs"
use_column_value => true
tracking_column_type => "numeric"
lowercase_column_names => false
schedule => "*/5 * * * * *"
statement => "SELECT *, UNIX_TIMESTAMP(updateTime) AS unix_ts_in_secs FROM shop_goods_es WHERE (UNIX_TIMESTAMP(updateTime)) > :sql_last_value AND updateTime < NOW() ORDER BY updateTime desc"
}
jdbc {
type => "cm2" #定义客户2的数据源
jdbc_driver_library => "/usr/local/logstash/java/mysql-connector-java-8.0.13.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_connection_string => "xxxx" #修改数据库地址
jdbc_user => "xxxx"
jdbc_password => "xxxx"
# 开启分页
jdbc_paging_enabled => true
tracking_column => "unix_ts_in_secs"
use_column_value => true
tracking_column_type => "numeric"
lowercase_column_names => false
schedule => "*/5 * * * * *"
statement => "SELECT *, UNIX_TIMESTAMP(updateTime) AS unix_ts_in_secs FROM shop_goods_es WHERE (UNIX_TIMESTAMP(updateTime)) > :sql_last_value AND updateTime < NOW() ORDER BY updateTime desc"
}
}
output{
if [type] == "cm1" { #判断输出地址，当type=cm1，的时候输出到客户1的ES索引
stdout{codec =>rubydebug}
elasticsearch {
hosts => ["xxxxx"]
index => "mall_goods_index"
document_id => "%{id}"
ssl => true
ilm_enabled => false
template_overwrite => true
# flush_size => 1000
template => "mall_goods_index.json" #使用自定义的索引格式
template_name => "mall_goods_index*"
}
}
if [type] == "cm2" { #判断输出地址，当type=cm2，的时候输出到客户2的ES索引
stdout{codec =>rubydebug}
elasticsearch {
hosts => ["xxxx"]
index => "test_goods_index"
document_id => "%{id}"
ssl => true
ilm_enabled => false
template_overwrite => true
# flush_size => 1000
template => "test_goods_index.json" #使用自定义的索引格式
template_name => "test_goods_index*"
}
}
}
启动logstash
/usr/local/logstash/bin/logstash
后续还有许多操作需要做，例如nohup挂载后台执行，定时任务清理日志，编写脚本检查自动重启logstash
|