使用logstash同步mysql 到es

下载安装Logstash
然后编写Logstash配置文件logstash.conf,注意mysql表一定要有主键，而且有更新时间字段
例如：

input {jdbc {# 设置 MySql/MariaDB 数据库url以及数据库名称jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/mydb?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true"# 用户名和密码jdbc_user => "root"jdbc_password => ""# 数据库驱动所在位置，可以是绝对路径或者相对路径jdbc_driver_library => "/usr/share/logstash/bin/mysql-connector-java-8.0.23.jar"# 驱动类名jdbc_driver_class => "com.mysql.cj.jdbc.Driver"# 开启分页jdbc_paging_enabled => "true"# 分页每页数量，可以自定义jdbc_page_size => "30"# 执行的sql文件路径statement => "SELECT *,FLOOR(addtime/86400) AS day,FLOOR(addtime/ 3600) AS hour FROM article WHERE addtime > :sql_last_value ORDER BY addtime ASC"# 设置定时任务间隔  含义：分、时、天、月、，全部为*默认含义为每分钟跑一次任务schedule => "* * * * *"# 索引类型type => "_doc"# 是否开启记录上次追踪的结果，也就是上次更新的时间，这个会记录到 last_run_metadata_path 的文件use_column_value => true# 记录上一次追踪的结果值last_run_metadata_path => "/opt/logstash/mysql_update_time"# 如果 use_column_value 为true， 配置本参数，追踪的 column 名，可以是自增id或者时间tracking_column => "addtime"# tracking_column 对应字段的类型tracking_column_type => "timestamp"# 是否清除 last_run_metadata_path 的记录，true则每次都从头开始查询所有的数据库记录clean_run => false# 数据库字段名称大写转小写lowercase_column_names => false}
}
output {elasticsearch {# 如果不配置type，动态模板会创建失败#document_type => "_doc"hosts => ["127.0.0.1"]#账号user => "elastic"password => ""index => "my_index"# 设置_docID和数据相同document_id => "%{id}"data_stream =>falsetemplate => "/opt/logstash/template.json"#是否重写模版template_overwrite => falsemanage_template => false}# 日志输出stdout {codec => json_lines}
}

编写对应的模板文件/opt/logstash/template.json，这里注意和ES索引的mapping文件字段格式上可能有些差别。

{"index_patterns": ["my_index"],"template": {"settings": {"index": {"number_of_shards": 3,"number_of_replicas": 2,}},"mappings": {"properties": {"id": {"type": "keyword"},"addtime": {"type": "date","format": "epoch_millis"},"day": {"type": "long"},"hour": {"type": "long"},"title": {"type": "text","analyzer": "ik_smart","search_analyzer": "ik_smart"},}}}
}