标签：doc rem 移除 timestamp hosts conf 17. shang 选择
wget https://artifacts.elastic.co/downloads/logstash/logstash-6.5.2.zip
unzip logstash-6.5.2.zip && cd logstash-6.5.2
bin/logstash-plugin install logstash-input-jdbc
bin/logstash-plugin install logstash-output-elasticsearch
wget http://central.maven.org/maven2/mysql/mysql-connector-java/5.1.47/mysql-connector-java-5.1.47.jar
unzip mysql-connector-java-5.1.47.zip
使用 logstash-input-jdbc 插件读取 mysql 的数据，这个插件的工作原理比较简单，就是定时执行一个 sql，然后将 sql 执行的结果写入到流中，增量获取的方式没有通过 binlog 方式同步，而是用一个递增字段作为条件去查询，每次都记录当前查询的位置，由于递增的特性，只需要查询比当前大的记录即可获取这段时间内的全部增量，一般的递增字段有两种，AUTO_INCREMENT 的主键 id 和 ON UPDATE CURRENT_TIMESTAMP 的 update_time 字段，id 字段只适用于那种只有插入没有更新的表，update_time 更加通用一些，建议在 mysql 表设计的时候都增加一个 update_time 字段
jdbc_driver_library: jdbc mysql 驱动的路径，在上一步中已经下载jdbc_driver_class: 驱动类的名字，mysql 填 com.mysql.jdbc.Driver 就好了jdbc_connection_string: mysql 地址jdbc_user: mysql 用户jdbc_password: mysql 密码schedule: 执行 sql 时机，类似 crontab 的调度statement: 要执行的 sql，以 ":" 开头是定义的变量，可以通过 parameters 来设置变量，这里的 sql_last_value是内置的变量，表示上一次 sql 执行中 update_time 的值，这里 update_time 条件是 >= 因为时间有可能相等，没有等号可能会漏掉一些增量use_column_value: 使用递增列的值tracking_column_type: 递增字段的类型，numeric 表示数值类型, timestamp 表示时间戳类型tracking_column: 递增字段的名称，这里使用 update_time 这一列，这列的类型是 timestamplast_run_metadata_path: 同步点文件，这个文件记录了上次的同步点，重启时会读取这个文件，这个文件可以手动修改input { jdbc { jdbc_driver_library => "../mysql-connector-java-5.1.46/mysql-connector-java-5.1.46-bin.jar" jdbc_driver_class => "com.mysql.jdbc.Driver" jdbc_connection_string => "jdbc:mysql://<mysql_host>:3306/rta" jdbc_user => "<username>" jdbc_password => "<password>" schedule => "* * * * *" statement => "SELECT * FROM table WHERE update_time >= :sql_last_value" use_column_value => true tracking_column_type => "timestamp" tracking_column => "update_time" last_run_metadata_path => "syncpoint_table" } }
hosts: es 集群地址user: es 用户名password: es 密码index: 导入到 es 中的 index 名，这里我直接设置成了 mysql 表的名字document_id: 导入到 es 中的文档 id，这个需要设置成主键，否则同一条记录更新后在 es 中会出现两条记录，%{id} 表示引用 mysql 表中 id 字段的值output { elasticsearch { hosts => ["172.31.22.165", "172.31.17.241", "172.31.30.84", "172.31.18.178"] user => "<user>" password => "<password>" index => "table" document_id => "%{id}" } }
把上面的代码保存到一个配置文件里面 sync_table.cfg，执行下面命令即可
bin/logstash -f config/sync_table.cfg
#logstash的配置文件需要自己创建，一下是我自己的配置文件
cat mysql_conf/mysql.conf
input {
stdin {
}
jdbc {
#数据库地址
jdbc_connection_string => "jdbc:mysql://localhost:3601/user"
jdbc_user => "admin"
jdbc_password => "passwd"
#数据库驱动路径
jdbc_driver_library => "/soft/logstash-5.6.9/mysql-connector-java-5.1.46/mysql-connector-java-5.1.46-bin.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
#sql路径，也可以是直接指定sql语句，字段需要换成 statement =>
statement_filepath => "/soft/logstash-5.6.9/pay_test.sql"
#是否开启记录追踪
record_last_run => "true"
#是否需要追踪字段，如果为true，则需要指定tracking_column，默认是timestamp
use_column_value => "true"
#指定追踪的字段，这里需要注意的是，建议选择主键字段，如果选择日期需要为实时表
tracking_column => "id"
#追踪字段的类型，目前只有数字和时间类型，默认是数字类型
#tracking_column_type => "number"
#设置时区
jdbc_default_timezone =>"Asia/Shanghai"
#是否每次清除last_run_metadata_path的内容
clean_run => "false"
#这里可以手动设置:sql_last_value的值，默认时间是1970-01-01，默认数字是0
last_run_metadata_path => "/soft/logstash-5.6.9/logstash_jdbc_last_run"
#多久同步一次
schedule => "*/5 * * * *"
#是否分页
jdbc_paging_enabled => "true"
jdbc_page_size => "50000"
}
}
filter {
#jdbc默认json，暂时没找到修改方法
json {
source => "message"
remove_field => ["message"]
}
mutate { #需要移除的字段
remove_field => "@timestamp"
remove_field => "type"
remove_field => "@version"
}
}
output {
elasticsearch {
hosts => "localhost:9200" #elasticsearch地址
index => "user" #elasticsearch索引
document_id => "%{id}" #elasticsearch的id，该值需要唯一，如果不唯一就不要加这个字段，默认生成
document_type => "log" #elasticsearch的type
}
}
标签：doc rem 移除 timestamp hosts conf 17. shang 选择
原文地址：https://www.cnblogs.com/FlyBlueSky/p/10097101.html
