logstash mysql 同步到 elasticsearch
es,kinana环境参考:https://blog.csdn.net/qq_38270106/article/details/88394869es集群环境参考:https://blog.csdn.net/qq_38270106/article/details/88411221Logstash是一个开源数据收集引擎,具有实时管道功能。Logstash可以动态地将来自不同数据源的数据统一起来...
es,kinana环境参考:https://blog.csdn.net/qq_38270106/article/details/88394869
es集群环境参考:https://blog.csdn.net/qq_38270106/article/details/88411221
Logstash是一个开源数据收集引擎,具有实时管道功能。Logstash可以动态地将来自不同数据源的数据统一起来,并将数据标准化到你所选择的目的地
1.上传logstash-6.4.3.tar.gz到服务中
2.tar –zxvf logstash-6.4.3.tar.gz
3.cd logstash-6.4.3
4. bin/logstash-plugin install logstash-input-jdbc
5. bin/logstash-plugin install logstash-output-elasticsearch
6. vi mysql.conf
input {
jdbc {
jdbc_driver_library => "/usr/local/sql/mysql-connector-java-5.1.46.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_connection_string => "jdbc:mysql://192.168.1.103:3306/test_elk"
jdbc_user => "root"
jdbc_password => "root"
schedule => "* * * * *"
statement => "SELECT * FROM user WHERE update_time >= :sql_last_value"
use_column_value => true
tracking_column_type => "timestamp"
tracking_column => "update_time"
last_run_metadata_path => "syncpoint_table"
}
}
output {
elasticsearch {
# ES的IP地址及端口
hosts => ["192.168.128.130:9200", "192.168.128.130:9201"]
# 索引名称 可自定义
index => "user"
# 需要关联的数据库中有有一个id字段,对应类型中的id
document_id => "%{id}"
document_type => "user"
}
stdout {
# JSON格式输出
codec => json_lines
}
}
说明:
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: 同步点文件,这个文件记录了上次的同步点,重启时会读取这个文件,这个文件可以手动修改
7. 创建user表
CREATE TABLE `user` (
`id` bigint(20) NOT NULL AUTO_INCREMENT,
`name` varchar(255) NOT NULL,
`update_time` timestamp NULL DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
8.上传mysql jar到/usr/local/sql/ mysql-connector-java-5.1.46.jar
9. 启动logstash
./bin/logstash -f mysql.conf
数据库插入两条记录:
同步log打印:
kinana查看es的数据:
多文件方式同步ES数据
一个 logstash 实例可以借助 pipelines 机制同步多个表,只需要写多个配置文件就可以了,假设我们有两个表 table1 和 table2,对应两个配置文件 mysql_user.conf 和 mysql_member.conf,
1. 创建member表
CREATE TABLE `member` (
`id` bigint(20) NOT NULL AUTO_INCREMENT,
`name` varchar(255) NOT NULL,
`update_time` timestamp NULL DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
2.上传到 /usr/local/sql目录下
mysql_user.conf
input {
jdbc {
jdbc_driver_library => "/usr/local/sql/mysql-connector-java-5.1.46.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_connection_string => "jdbc:mysql://192.168.1.103:3306/test_elk"
jdbc_user => "root"
jdbc_password => "root"
schedule => "* * * * *"
statement => "SELECT * FROM user WHERE update_time >= :sql_last_value"
use_column_value => true
tracking_column_type => "timestamp"
tracking_column => "update_time"
last_run_metadata_path => "syncpoint_table"
}
}
output {
elasticsearch {
# ES的IP地址及端口
hosts => ["192.168.128.130:9200", "192.168.128.130:9201"]
# 索引名称 可自定义
index => "user"
# 需要关联的数据库中有有一个id字段,对应类型中的id
document_id => "%{id}"
document_type => "user"
}
stdout {
# JSON格式输出
codec => json_lines
}
}
mysql_member.conf
input {
jdbc {
jdbc_driver_library => "/usr/local/sql/mysql-connector-java-5.1.46.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_connection_string => "jdbc:mysql://192.168.1.103:3306/test_elk"
jdbc_user => "root"
jdbc_password => "root"
schedule => "* * * * *"
statement => "SELECT * FROM member WHERE update_time >= :sql_last_value"
use_column_value => true
tracking_column_type => "timestamp"
tracking_column => "update_time"
last_run_metadata_path => "syncpoint_table"
}
}
output {
elasticsearch {
# ES的IP地址及端口
hosts => ["192.168.128.130:9200", "192.168.128.130:9201"]
# 索引名称 可自定义
index => "member"
# 需要关联的数据库中有有一个id字段,对应类型中的id
document_id => "%{id}"
document_type => "member"
}
stdout {
# JSON格式输出
codec => json_lines
}
}
只需要修改sql语句,es索引和文档类型
2.在 config/pipelines.yml 中配置
cd conf
vi pipelines.yml
- pipeline.id: table1
path.config: "/usr/local/sql/mysql_user.conf"
- pipeline.id: table2
path.config: "/usr/local/sql/mysql_member.conf"
3.启动
cd /usr/local/logstash-6.4.3
./bin/logstash
member表插入两条数据:
kinana查看es的同步数据
更多推荐
所有评论(0)