elasticsearch学习笔记2 - logstash jdbc 同步MYSQL到ES

logstash是一个管道工具 input 输入 output输出 filter过滤条件 咋们初级先了解到这些再说

比较优秀的教程文档 作为基础知识需要了解：

https://blog.csdn.net/qq_19283249/article/details/130839158?ops_request_misc=%257B%2522request%255Fid%2522%253A%2522170564710516800215045067%2522%252C%2522scm%2522%253A%252220140713.130102334..%2522%257D&request_id=170564710516800215045067&biz_id=0&utm_medium=distribute.pc_search_result.none-task-blog-2~all~top_click~default-2-130839158-null-null.142^v99^pc_search_result_base1&utm_term=logstash&spm=1018.2226.3001.4449

同步MySQL到es需要用到logstash的工具jdbc

1、需要MySQL jar包 版本 8.0.25  zip 压缩包 解压 jar 导入 ubuntu中使用

1、创建logstash  MySQL文件夹

mkdir mysql

将jar包放入mysql文件夹中

2：创建数据同步配置文件

input {
    jdbc {
        #注意mysql连接地址一定要用ip，不能使用localhost等
        jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/erp-dev?characterEncoding=utf8&serverTimezone=Asia/Shanghai"
        jdbc_user => "" 
        jdbc_password => ""
        
        #数据库重连尝试
        connection_retry_attempts => "3"
        
        #数据库连接可用校验超时时间，默认为3600s
        jdbc_validation_timeout => "3600"
        
        #这个jar包的地址是容器内的地址
        jdbc_driver_library => "/usr/bin/lib/es/logstash-8.11.0/mysql/mysql-connector-java-8.0.25.jar"
        jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
        
        #开启分页查询（默认是false）
        jdbc_paging_enabled => "true"
        
        #单次分页查询条数（默认100000，字段较多的话，可以适当调整这个数值）
        jdbc_page_size => "500"
        
        #执行的sql语句
    statement_filepath => "/usr/bin/lib/es/logstash-8.11.0/heyines/order.sql"
    }
}

filter {
  mutate {
     copy => { "id" => "[@metadata][_id]"}
     remove_field => ["@version","@timestamp"] #remove system add columns 
    }   
} 


output {
     elasticsearch {
        hosts => ["127.0.0.1:9200"]                            # Elasticsearc主机地址
        index => "t_es_order"                                      # ES索引名称
        document_id => "%{id}"                                # 使用MySQL中的ID作为文档ID
     }
}

jdbc 同步遇到的问题

1、同步字段时间提前8小时

jdbc_connection_string  中 设置时间时区为上海  serverTimezone=Asia/Shanghai

2、同步时间字段 格式不正确 无法再fliter中修改时间格式

只能修改查询SQL语句

datetime类型字段 两种

,SUBSTRING(DATE_FORMAT(t0.`timestamp`,'%Y-%m-%d %H:%i:%S.%f'),1,23) as `timestamp`
,DATE_FORMAT(t0.`create_time`,'%Y-%m-%d %H:%i:%S')as `create_time`