安装logstash和es这些看之前的一篇文章
第一步修改logstash.conf文件
input {
jdbc {
jdbc_driver_library => "/usr/share/logstash/lib/mysql-connector-j-8.0.33.jar"
jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
jdbc_connection_string => "jdbc:mysql://192.168.150.7:3306/data-synchronism"
jdbc_user => "root"
jdbc_password => "root"
#启用追踪，如果为true，则需要指定tracking_column
use_column_value => true
#指定追踪的字段，
tracking_column => "update_time"
#追踪字段的类型，目前只有数字(numeric)和时间类型(timestamp)，默认是数字类型
tracking_column_type => "timestamp"
# 记录最后一次运行的结果
record_last_run => true
# 上面运行结果的保存位置 这个文件中保存着最后一次查询，update_time字段的值，下一次查询就可以用这个值来做where条件
last_run_metadata_path => "jdbc-position.txt"
# 因为我上面使用的是timestamp追踪字段类型，sql_last_value这里开始的值是1970-01-01 00:00:00
# 之后sql_last_value 的值就是最后一次查询 追踪字段的值，这个值保存在上方定义的 jdbc-position.txt文件中
statement => "SELECT * from demo where update_time >:sql_last_value"
# cron 定时 每秒执行一次
schedule => " * * * * * *"
}
}
# filter {
# # 添加同步前的时间戳（这里用事件进入filter阶段的时间作为示例）
# ruby {
# code => "event.set('sync_before_time', Time.now.utc.iso8601)"
# }
# # ... 其他filter插件和代码 ...
# # 在filter阶段末尾或适当位置添加同步完的时间戳（这里为了示例简单，仍然用当前时间）
# # 注意：在实际应用中，这里的时间可能并不完全代表数据真正写入目标数据源的时间，
# # 因为output阶段还可能存在延迟。因此，这个字段更多是作为逻辑上的“同步完”时间。
# ruby {
# code => "event.set('sync_after_time', Time.now.utc.iso8601)"
# }
# }
output {
elasticsearch {
hosts => ["http://es1:9200", "http://es2:9200", "http://es3:9200"]
# 索引：等于数据的名称
index => "demo"
# 文档id
document_id => "%{id}" # 指定文档id， %{}取字段中的值 id为字段名
user => "elastic"
password => "elastic"
}
# 控制台也输出看看
stdout {
codec => rubydebug
# { metadata => true }
}
}
上面配置自己的MySQL数据库地址
第二步创建数据库并造100万数据
create database `data-synchronism`;
use `data-synchronism`;
CREATE TABLE demo (
id INT AUTO_INCREMENT PRIMARY KEY,
username VARCHAR(50) NOT NULL,
email VARCHAR(100) NOT NULL,
create_time datetime,
update_time datetime
);
DELIMITER $$
CREATE PROCEDURE generate_sample_data()
BEGIN
DECLARE v_counter INT DEFAULT 0;
WHILE v_counter < 1000000 DO
INSERT INTO demo (username, email, create_time, update_time) VALUES (
CONCAT('user', v_counter),
CONCAT('user', v_counter, '@example.com'),
NOW(),
NOW()
);
SET v_counter = v_counter + 1;
END WHILE;
END$$
DELIMITER ;
CALL generate_sample_data();
导入的时间会很长耐心等待
mysqlconnectorj8033jarmy_elklogstashlib_97">第三步将mysql-connector-j-8.0.33.jar上传到/my_elk/logstash/lib中
第四步通过docker-compose启动容器
docker-compose -f es.yml up -d
启动后马上把MySQL的驱动包复制到容器中
docker cp /my_elk/logstash/lib/mysql-connector-j-8.0.33.jar logstash:/usr/share/logstash/lib
最后重启一下logstash
docker restart logstash
最后验证同步数据
通过日志查看正在同步的数据
通过kibana查询es中的数据
同步100万数据的时间
容器创建的时间跟系统时间差了八小时，通过测试可以得出结果，因容器启动也需要时间，所以同步100万数据的大概时间是20分钟左右。
由于是本地测试，网络波动不是很大，如果是远程服务器测试，需要的时间大约是30分钟左右
同步500多万数据的时间
上面同步100万数据容器是冷启动的，这次500多万数据容器是热启动时间快了很多
