当前位置：首页 > article >正文

MySQL数据高效同步到Elasticsearch的四大方案

article 2025/3/21 1:14:39

引言

一、为什么需要MySQL到ES的同步？

二、四大同步方案对比

三、方案详解与代码实战

1. 应用层双写：简单但强耦合

2. 定时任务同步：可控的准实时

3. Logstash JDBC：离线迁移利器

4. Binlog监听：生产级实时同步（推荐）

四、避坑指南：关键注意事项

五、总结

引言

Elasticsearch（ES）凭借其强大的全文搜索和实时分析能力，已成为现代应用的核心组件。但当数据存储在MySQL时，如何实现高效、可靠的双向同步？本文将深入解析四种主流同步方案，涵盖从简单双写到生产级实时同步的全场景，并提供可落地的代码实现。

官网

一、为什么需要MySQL到ES的同步？

1. 全文搜索：ES支持分词、模糊匹配，弥补MySQL LIKE查询性能差的缺陷

2. 复杂聚合：ES Bucket和Metric聚合实现毫秒级多维分析

3. 数据异构：ES支持嵌套文档、向量搜索等灵活的数据结构

4. 读写分离：将复杂查询流量从MySQL卸载到ES，提升系统整体性能

二、四大同步方案对比

方案	实时性	数据一致性	开发成本	适用场景
应用层双写	实时	难保证	低	小型项目，数据量小
定时任务同步	分钟级	最终一致	中	允许延迟，增量同步场景
Logstash JDBC	小时级	最终一致	低	离线历史数据迁移
Binlog监听	秒级	强一致	高	生产环境高实时性要求

三、方案详解与代码实战

1. 应用层双写：简单但强耦合

原理：在业务代码中同步写入MySQL和ES，适合初创项目快速验证。

// Node.js 示例（注意事务回滚！）
async function createOrder(orderData) {
  // 1. MySQL写入
  const [mysqlResult] = await mysql.query(
    'INSERT INTO orders SET ?', orderData
  );
  
  // 2. ES同步
  try {
    await elasticClient.index({
      index: 'orders',
      id: mysqlResult.insertId.toString(),
      body: orderData
    });
  } catch (e) {
    // ES写入失败则回滚MySQL
    await mysql.query('DELETE FROM orders WHERE id = ?', [mysqlResult.insertId]);
    throw e;
  }
}

缺陷：

•业务侵入性强，需维护两套数据模型

•分布式事务难题（建议本地事务表+补偿机制）

2. 定时任务同步：可控的准实时

核心步骤：

1. MySQL表添加`updatedat`字段

2. 定时扫描增量数据批量推送到ES

// 使用Node.js定时任务（示例：每10分钟）
const schedule = require('node-schedule');
let lastSyncTime = new Date('2024-01-01');
schedule.scheduleJob('*/10 * * * *', async () => {
  const results = await mysql.query(
    `SELECT * FROM orders WHERE updated_at > ?`, 
    [lastSyncTime]
  );
  
  // 构造ES Bulk API请求体
  const bulkBody = results.flatMap(doc => [
    { index: { _index: 'orders', _id: doc.id } },
    { ...doc, timestamp: new Date() } // 可追加自定义字段
  ]);
  
  if (bulkBody.length > 0) {
    await elasticClient.bulk({ body: bulkBody });
    lastSyncTime = new Date(); // 持久化存储时间戳防宕机
  }
});

优化技巧：

•使用`trackingcolumn`记录断点（如Redis存储`lastSyncTime`）

•分页查询避免内存溢出

3. Logstash JDBC：离线迁移利器

配置要点：

•安装MySQL驱动到Logstash的`/logstash-core/lib/jars/`

•定时轮询策略

# mysql-to-es.conf
input {
  jdbc {
    jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://localhost:3306/mydb"
    jdbc_user => "admin"
    jdbc_password => "Passw0rd!"
    schedule => "*/30 * * * *" # 每30分钟
    statement => "SELECT * FROM products WHERE updated_at > :sql_last_value"
    tracking_column => "updated_at"
    tracking_column_type => "timestamp"
    last_run_metadata_path => "/tmp/products_last_run.time"
  }
}
output {
  elasticsearch {
    hosts => ["http://es-node1:9200"]
    index => "products"
    document_id => "%{id}"
  }
}

启动命令：

bin/logstash -f mysql-to-es.conf

4. Binlog监听：生产级实时同步（推荐）

架构：

`MySQL -> Canal/Debezium -> Kafka -> ES Consumer`

Debezium实战步骤：

1. 启动Kafka集群

docker-compose up -d zookeeper kafka schema-registry

2. 部署Debezium MySQL Connector

curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{
  "name": "mysql-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "mysql",
    "database.user": "debezium",
    "database.password": "dbz",
    "database.server.name": "inventory",
    "table.include.list": "inventory.products",
    "database.history.kafka.bootstrap.servers": "kafka:9092"
  }
}'

3. 编写ES消费者

const { Kafka } = require('kafkajs');
const kafka = new Kafka({ brokers: ['kafka:9092'] });
const consumer = kafka.consumer({ groupId: 'es-sync' });
consumer.connect().then(() => {
  consumer.subscribe({ topic: 'inventory.products' });
  consumer.run({
    eachMessage: async ({ message }) => {
      const event = JSON.parse(message.value);
      switch (event.op) {
        case 'c':
        case 'u':
          await esClient.index({
            index: 'products',
            id: event.after.id,
            body: event.after
          });
          break;
        case 'd':
          await esClient.delete({ index: 'products', id: event.before.id });
          break;
      }
    }
  });
});