- 下载插件
./bin/logstash-plugin install logstash-input-jdbc
./bin/logstash-plugin install logstash-output-elasticsearch
-
配置作业文件
创建一个Logstash作业配置文件,文件命名为logstash-mysql-es.conf
-
配置文件内容
- 这里 mysql 驱动文件需要自己手动下载,并复制到 logstash 目录下,默认路径为
/usr/share/logstash/logstash-core/lib/jars
, 如果另外目录需要在配置中配置绝对路径
input{ jdbc { jdbc_driver_library => "mysql-connector-java-5.1.44-bin.jar" jdbc_driver_class => "com.mysql.jdbc.Driver" jdbc_connection_string => "jdbc:mysql://localhost.com:3306/db_name?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true&useSSL=false&zeroDateTimeBehavior=convertToNull&serverTimezone=Asia/Shanghai" jdbc_user => "db_user" jdbc_password => "db_password" # 开启分页 jdbc_paging_enabled => "true" # 每页数量 jdbc_page_size => "1000" jdbc_default_timezone =>"Asia/Shanghai" # 配置数据刷新频率,如果所有都为 '*' ,表示每分钟刷新一次,这也是 Mysql 数据同步的最小频率。 schedule => "* * * * *" # 数据同步查询接口,注意 sql 语句不需要带 ‘;’ 号结尾 statement => "select * from test where updatetime > :sql_last_value" ## 除了上面这种直接执行sql语句的,还有指定sql文件,当然只能二选一 statement_filepath => "/home/test/test.sql" # 启用追踪,如果为 true ,则需要指定 tracking_column use_column_value => true # 指定递增字段 tracking_column => "updatetime" # 递增字段类型,目前只有数字(numeric)和时间类型(timestamp),默认是数字类型 tracking_column_type => "timestamp" # 记录最后一次运行的结果 record_last_run => true # 上面运行结果保存的位置 last_run_metadata_path => "./logstash_jdbc_last_run" } } output{ elasticsearch { hosts => "localhost:9200" user => "elastic" password => "es_password" index => "employee" document_id => "%{id}" } stdout { codec => json_lines } }
- 这里 mysql 驱动文件需要自己手动下载,并复制到 logstash 目录下,默认路径为
-
启动运行
/usr/share/logstash/bin/logstash --path.settings /etc/logstash -f /home/elk/logstash/conf/logstash-mysql-es.conf
-
后台启动
nohup /usr/share/logstash/bin/logstash --path.settings /etc/logstash -f /home/elk/logstash/conf/logstash-mysql-es.conf &