搜索客，搜索人自己的社区

社区日报第1688期 (2023-08-21)

社区日报 • yuebancanghai 发表了文章 • 0 个评论 • 1447 次浏览 • 2023-08-21 19:38 • 来自相关话题

这个是不是多余的代码

Elasticsearch • Charele 发起了问题 • 1 人关注 • 0 个回复 • 3311 次浏览 • 2023-08-19 13:10 • 来自相关话题

社区日报第1687期 (2023-08-18）

社区日报 • laoyang360 发表了文章 • 0 个评论 • 1440 次浏览 • 2023-08-18 22:18 • 来自相关话题

请教一个关于关注列表用户搜索的场景问题

贡献

Elasticsearch • zcc_vv 回复了问题 • 2 人关注 • 1 个回复 • 1790 次浏览 • 2023-08-18 18:53 • 来自相关话题

elasticsearch 缺乏足够的无分段虚拟地址空间，导致集群故障，请问有什么优化方案吗

贡献

Elasticsearch • FFFrp 回复了问题 • 3 人关注 • 2 个回复 • 2954 次浏览 • 2023-08-25 10:48 • 来自相关话题

比FST还要快的东东？

Elasticsearch • Charele 发起了问题 • 1 人关注 • 0 个回复 • 2429 次浏览 • 2023-08-17 14:05 • 来自相关话题

社区日报第1686期 (2023-08-17）

社区日报 • Se7en 发表了文章 • 0 个评论 • 1637 次浏览 • 2023-08-17 09:15 • 来自相关话题

使用 Logstash 同步 MySQL 到 Easysearch

Easysearch • yangmf2040 发表了文章 • 0 个评论 • 2289 次浏览 • 2023-08-17 00:49 • 来自相关话题

从 MySQL 同步数据到 ES 有多种方案，这次我们使用 ELK 技术栈中的 Logstash 来将数据从 MySQL 同步到 Easysearch 。

方案前提

MySQL 表记录必须有主键，比如 id 字段。通过该字段，可将 Easysearch 索引数据与 MySQL 表数据形成一对一映射关系，支持修改。
MySQL 表记录必须有时间字段，以支持增量同步。

如果上述条件具备，便可使用 logstash 定期同步新写入或修改后的数据到 Easysearch 中。

方案演示

版本信息

MySQL: 5.7
Logstash: 7.10.2
Easysearch: 1.5.0

MySQL 设置

创建演示用的表。
yaml CREATE DATABASE es_db; USE es_db; DROP TABLE IF EXISTS es_table; CREATE TABLE es_table ( id BIGINT(20) UNSIGNED NOT NULL, PRIMARY KEY (id), UNIQUE KEY unique_id (id), client_name VARCHAR(32) NOT NULL, modification_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, insertion_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ); 
说明
- id 字段: 主键、唯一键，将作为 Easysearch 索引中的 doc id 字段。
- modification_time 字段: 表记录的插入和修改都会记录在此。
- client_name: 代表用户数据。
- insertion_time: 可省略，用来记录数据插入到 MySQL 数据的时间。
 
 插入数据
 
 yaml INSERT INTO es_table (id, client_name) VALUES (1, 'test 1'); INSERT INTO es_table (id, client_name) VALUES (2, 'test 2'); INSERT INTO es_table (id, client_name) VALUES (3, 'test 3'); 
 
 Logstash
 
 配置文件
 yaml input { jdbc { jdbc_driver_library => "./mysql-connector-j-8.1.0/mysql-connector-j-8.1.0.jar" jdbc_driver_class => "com.mysql.jdbc.Driver" jdbc_connection_string => "jdbc:mysql://192.168.56.3:3306/es_db" jdbc_user => "root" jdbc_password => "password" jdbc_paging_enabled => true tracking_column => "unix_ts_in_secs" use_column_value => true tracking_column_type => "numeric" last_run_metadata_path => "./.mysql-es_table-sql_last_value.yml" schedule => "*/5 * * * * *" statement => "SELECT *, UNIX_TIMESTAMP(modification_time) AS unix_ts_in_secs FROM es_table WHERE (UNIX_TIMESTAMP(modification_time) > :sql_last_value AND modification_time < NOW()) ORDER BY modification_time ASC" } jdbc { jdbc_driver_library => "./mysql-connector-j-8.1.0/mysql-connector-j-8.1.0.jar" jdbc_driver_class => "com.mysql.jdbc.Driver" jdbc_connection_string => "jdbc:mysql://192.168.56.3:3306/es_db" jdbc_user => "root" jdbc_password => "password" schedule => "*/5 * * * * *" statement => "SELECT count(*) AS count,'es_table' AS table_name from es_table" } } filter { if ![table_name] { mutate { copy => { "id" => "[@metadata][_id]"} remove_field => ["@version", "unix_ts_in_secs","@timestamp"] add_field => { "[@metadata][target_index]" => "mysql_es_table" } } } else { mutate { add_field => { "[@metadata][target_index]" => "table_counts" } remove_field => ["@version"] } uuid { target => "[@metadata][_id]" overwrite => true } } } output { elasticsearch { hosts => ["<a href="https://localhost:9200"" rel="nofollow" target="_blank">https://localhost:9200"</a>] user => "admin" password => "f0c6fc61fe5f7b084c00" ssl_certificate_verification => "false" index => "%{[@metadata][target_index]}" manage_template => "false" document_id => "%{[@metadata][_id]}" } } 
- 每 5 秒钟同步一次 es_table 表的数据到 mysql_sync_idx 索引。
- 每 5 秒统计一次 es_table 表的记录条数到 table_counts 索引，用于监控。
 
 启动 logstash
 
 yaml ./bin/logstash -f sync_es_table.conf 
 查看同步结果, 3 条数据都已同步到索引。
 ![](https://www.infinilabs.com/img ... /1.png)
 Mysql 数据库新增记录
 yaml INSERT INTO es_table (id, client_name) VALUES (4, 'test 4'); 
 Easysearch 确认新增
 ![](https://www.infinilabs.com/img ... /2.png)
 
 Mysql 数据库修改记录
 yaml UPDATE es_table SET client_name = 'test 0001' WHERE id=1; 
 Easysearch 确认修改
 ![](https://www.infinilabs.com/img ... /3.png)
 
 删除数据
 
 Logstash 无法直接删除操作到 ES ，有两个方案：
在表中增加 is_deleted 字段，实现软删除，可达到同步的目的。查询过滤掉 is_deleted : true 的记录，后续通过脚本等方式定期清理 is_deleted : true 的数据。
执行删除操作的程序，删除完 MySQL 中的记录后，继续删除 Easysearch 中的记录。

同步监控

数据已经在 ES 中了，我们可利用 INFINI Console 的数据看板来监控数据是否同步，展示表记录数、索引记录数及其变化。
![](https://www.infinilabs.com/img ... /4.png)