Logstash同步Mysql至ES

程序推送商品信息至Elasticsearch,因为同步的问题，导致了商品信息不同步

使用Logstash同步Mysql数据库数据

• jdk8
• logstash 7.7.1
• Elasticsearch
• mysql-connector-java-8.0.13.jar

# 安装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)

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/pipeline5 * * * * *"
         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*"

      }
 }
}

/usr/local/logstash/bin/logstash

后续还有许多操作需要做，例如nohup挂载后台执行，定时任务清理日志，编写脚本检查自动重启logstash