MySQL | 实战 | 4 种将数据同步到ES方案

在现代应用中，使用MySQL作为数据库和Elasticsearch（ES）作为搜索引擎的组合十分常见。MySQL适合进行结构化数据的存储与管理，而Elasticsearch则在全文搜索和复杂查询上具有显著优势。为了将MySQL中的数据高效同步到Elasticsearch中，开发者可以选择多种方案。以下是四种常见的同步方案：

1. 数据库触发器 + 消息队列

使用数据库触发器可以在数据插入、更新或删除时触发相应的操作。通过触发器，将变化记录到消息队列（如Kafka、RabbitMQ），然后由消费者从消息队列中获取数据并同步到Elasticsearch中。

示例代码：

-- 创建触发器
DELIMITER //

CREATE TRIGGER after_insert_user
AFTER INSERT ON users
FOR EACH ROW 
BEGIN
   INSERT INTO user_changes (user_id, operation) VALUES (NEW.id, 'INSERT');
END; //

DELIMITER ;

消费者可以使用Python的Kafka客户端从消息队列中读取数据并将其插入到ES中：

from kafka import KafkaConsumer
from elasticsearch import Elasticsearch

es = Elasticsearch(["http://localhost:9200"])
consumer = KafkaConsumer('user_changes', bootstrap_servers='localhost:9092')

for message in consumer:
    data = message.value
    es.index(index='users', id=data['user_id'], body=data)

2. MySQL Binlog + Logstash

MySQL的二进制日志（Binlog）记录了所有更改数据库的操作。使用Logstash可以解析这个Binlog，并将数据同步到Elasticsearch。

配置示例：

首先，确保MySQL启用了Binlog。然后在Logstash中配置input插件：

input {
  jdbc {
    jdbc_driver_library => "mysql-connector-java.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://localhost:3306/your_database"
    jdbc_user => "your_username"
    jdbc_password => "your_password"
    statement => "SELECT * FROM users WHERE updated_at > :sql_last_value"
    use_column_value => true
    tracking_column => "updated_at"
    clean_run => false
  }
}

然后配置output插件，将数据发送到Elasticsearch：

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "users"
    document_id => "%{id}"
  }
}

3. 应用层同步

在你的应用程序代码中，处理数据的插入、更新和删除操作时，可以同时更新MySQL和Elasticsearch。这种方法比较简单直接，但需要谨慎处理错误和异常，确保数据一致性。

示例代码：

import mysql.connector
from elasticsearch import Elasticsearch

def add_user(user_data):
    # MySQL连接
    db = mysql.connector.connect(host="localhost", user="your_username", password="your_password", database="your_database")
    cursor = db.cursor()

    # 插入MySQL
    cursor.execute("INSERT INTO users (name, email) VALUES (%s, %s)", (user_data['name'], user_data['email']))
    db.commit()

    # 更新Elasticsearch
    es = Elasticsearch(["http://localhost:9200"])
    es.index(index='users', id=cursor.lastrowid, body=user_data)

    cursor.close()
    db.close()

4. 定时批量同步

如果对实时性要求不高，可以选择定时批量将数据从MySQL同步到Elasticsearch。这种方法通常使用定时任务（如Cron）运行脚本，将需要的数据查询出来并更新到Elasticsearch中。

示例代码：

import mysql.connector
from elasticsearch import Elasticsearch

def sync_users():
    # MySQL连接
    db = mysql.connector.connect(host="localhost", user="your_username", password="your_password", database="your_database")
    cursor = db.cursor(dictionary=True)

    # 查询所有用户
    cursor.execute("SELECT * FROM users")
    users = cursor.fetchall()

    # 更新Elasticsearch
    es = Elasticsearch(["http://localhost:9200"])
    for user in users:
        es.index(index='users', id=user['id'], body=user)

    cursor.close()
    db.close()

# 设置每小时执行一次
import schedule
import time

schedule.every().hour.do(sync_users)

while True:
    schedule.run_pending()
    time.sleep(1)

总结

以上四种方法各有优缺点，开发者可以根据实际业务场景、数据量、实时性要求等因素选择合适的方案。无论选择哪种方式，都需要关注数据同步的一致性和完整性，确保用户提供的搜索体验流畅高效。