使用 Logstash 同步 MySQL 到 Easysearch

从 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 设置 #

创建演示用的表。

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 数据的时间。

插入数据 #

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 #

配置文件

展开查看配置文件 ...

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 {
#  stdout { codec =>  rubydebug { metadata => true } }
  elasticsearch {
      hosts => ["https://localhost:9200"]
      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 #

./bin/logstash -f sync_es_table.conf

查看同步结果, 3 条数据都已同步到索引。

Mysql 数据库新增记录

INSERT INTO es_table (id, client_name) VALUES (4, 'test 4');

Easysearch 确认新增

Mysql 数据库修改记录

UPDATE es_table SET client_name = 'test 0001' WHERE id=1;

Easysearch 确认修改

删除数据 #

Logstash 无法直接删除操作到 ES ，有两个方案：

在表中增加 is_deleted 字段，实现软删除，可达到同步的目的。查询过滤掉 is_deleted : true 的记录，后续通过脚本等方式定期清理 is_deleted : true 的数据。
执行删除操作的程序，删除完 MySQL 中的记录后，继续删除 Easysearch 中的记录。

同步监控 #

数据已经在 ES 中了，我们可利用 INFINI Console 的数据看板来监控数据是否同步，展示表记录数、索引记录数及其变化。

杨帆

追求极致，无限可能。

· TDBC 2025 大会聚焦 AI 与数据库融合，极限科技发布新一代 Coco AI 搜索平台 · 极限科技 Coco AI 荣获 2025 首届人工智能应用创新大赛全国一等奖 · 搜索型数据库的技术发展历程与趋势前瞻 · 极限网关助力好未来 Elasticsearch 容器化升级 · 国产搜索引擎崛起：Elasticsearch 国产化加速

· 极限科技开源赞助计划：欢迎开源项目申请支持 · 探索低空经济“关键一跃” —— 技术如何驱动商业落地线上研讨会圆满举办 · 极限科技荣膺 2025 金猿奖 — “年度国产化优秀代表厂商”，自主可控搜索方案 Easysearch 获行业高度认可 · 极限科技 Coco AI 荣获 2025 IT168 技术卓越奖 - 创新产品奖 · APM（三）：监控 Python 服务链 · INFINI Labs 产品更新 | Coco AI v0.10 × Easysearch v2.0 联袂上线：UI 全面重构，体验焕然一新 · Easy-Es 2.1.0-easysearch 版本发布 · APM（二）：监控 Python 服务 · 🔥 程序员爆哭！我们让 COCO AI 接管 GitLab 审查后，团队直接起飞：连 CTO 都说“这玩意儿比人靠谱多了” · APM（一）: Skywalking 与 Easyearch 集成

2026 x

开源 x

赞助 x

开源生态 x

社区 x

低空经济 x

商业化 x

Easysearch x

数据分析 x

金猿奖 x

国产化 x

搜索引擎 x

Coco AI x

技术卓越奖 x

创新产品奖 x

IT168 x

APM x

Skywalking x

产品更新 x

Easy-Es x

Coco x

AI x

GitLab x

代码审核 x

人工智能 x

石油石化 x

performance x

Gitee x

投票 x

Meilisearch x

Rust x

轻量级 x

搜索百科 x

Docker x

Docker Compose x

Easyserach x

Console x

DevOps x

Elasticsearch x

国产替代 x

backup x

snapshot x

CCR x

Gateway x

esdump x

source_reuse x

ignore_above x

OpenSearch x

AWS x

Lucene x

Solr x

Easyearch x

发明专利 x

数据分区 x

国际专利 x

一等奖 x

人工智能应用创新大赛 x

bulk x

embedding x

OpenAI x

IK x

TDBC x

2025 x

信通院 x

可信数据库大会 x

搜索型数据库 x

中国数据库产业图谱 x

上海开源创新菁英荟 x

开源创新新星企业 x

Workshop x

AI 搜索 x

智能助手 x

Automation x

Logstash x

MongoDB x

开源中国 x

直播 x

merge x

Elasticsearch 9 x

GitCode x

AI搜索 x

Cloud x

rollup x

Kubernetes x

Operator x

Arm64 x

Snapshot x

S3 x

Grafana x

Opensearch x

Nginx x

直播活动 x

搜索客社区 x

Meetup x

ES x

企业搜索 x

DeepSeek x

RAG x

certificate x

windows x

Rollup x

TopN x

Filebeat x

Ubuntu x

请求限速 x

INFINI Console x

指标 x

Kibana x

多集群 x

client x

Spring Boot x

ECE x

ES Bulk x

vector database x

Postgres x

可搜索快照 x

SDK x

官网 x

Web 开发 x

Next.js x

React x

Three.js x

Metrics x

Helm x

filter x

querycache x

practice x

Agent x

localStorage x

响应式 x

时间组件 x

时区组件 x

极限科技 x

三周年 x

周年庆 x

国家高新技术企业 x

校园招聘 x

湖北工业大学 x

Tauri x

Web 开发人员 x

桌面应用开发 x

桌面端 x

Electron x

Pizza x

认证培训 x

报名 x

Scrapy x

爬虫 x

Rust开发者大会 x

docsearch x

文档搜索 x

Easyseach x

有奖征文 x

黑神话悟空 x

EKS x

征文系列 x

跨集群搜索 x

科技中小企业 x

白皮书 x

Python SDK x

数据库产业图谱 x

超大规模 x

分布式集群 x

写入限流 x

2024可信数据库发展大会 x

创新型中小企业 x

搜索数据库 x

正排索引 x

免费许可证 x

K8S x

DTC2024 x

实时搜索 x

ES国产化 x

Redis x

OOM x

测试 x

内存 x

趋势 x

AI绘画 x

Stable Diffusion x

Diffusion x

Model x

GAN x

语义搜索 x

知识图 x

向量数据库 x

中国信通院 x

星河（Galaxy） x

标杆案例 x

鲲鹏 x

鲲鹏技术认证 x

客户端 x

日志平台 x

LDAP x

Loadgen x

中国一汽 x

国内数据库 x

墨天轮 x

监控系统 x

集成测试 x

ZSTD x

Helm Charts x

国产适配 x

兆芯 x

Linux x

LoongArch x

信创适配 x

二维拆分算法 x

中国移动云 x

Vault x

加密 x

安全工具 x

kNN x

向量检索 x

图片搜索 x

Alerting x

SQL x

搜索 x

Embedding x

可信数据库 x

统信 x

海光 x

龙芯 x

restore x

Arm x

大数据企业证书 x

移动云大会 x

信通院产品评测 x

国内首家 x

数据可视化 x

北京软协 x

第十届理事会会员单位 x

Apache Arrow x

宣传片 x

大会分享 x

多集群管理 x

无缝数据迁移 x

Loadrun x

INFINI Gateway x

log4j x