logstash 获取
wget https://artifacts.elastic.co/downloads/logstash/logstash-6.5.2.zip unzip logstash-6.5.2.zip && cd logstash-6.5.2
安装 jdbc 和 elasticsearch 插件
bin/logstash-plugin install logstash-input-jdbc bin/logstash-plugin install logstash-output-elasticsearch
获取 jdbc mysql 驱动
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
使用 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 这一列,这列的类型是timestamp
last_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" } }
logstash-output-elasticsearch
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 } }
原文地址:https://www.cnblogs.com/FlyBlueSky/p/10097101.html
时间: 2024-10-07 22:09:21