如何高效实现 MySQL 与 elasticsearch 的数据同步

原数据库的同步问题

由于传统的 mysql 数据库并不擅长海量数据的检索，当数据量到达一定规模时（估算单表两千万左右），查询和插入的耗时会明显增加。同样，当需要对这些数据进行模糊查询或是数据分析时，MySQL作为事务型关系数据库很难提供良好的性能支持。使用适合的数据库来实现模糊查询是解决这个问题的关键。

那具体该如何实现呢？在又拍云以往的项目中，也有遇到相似的问题。之前采用的方法是在业务中编写代码，然后同步到 elasticsearch 中。具体是这样实施的：每个系统编写特定的代码，修改 MySQL 数据库后，再将更新的数据直接推送到需要同步的数据库中，或推送到队列由消费程序来写入到数据库中。

系统高耦合，侵入式代码，使得业务逻辑复杂度增加

工作量和复杂度增加

解决思路及方案

调整架构

既然以往的方案有明显的缺点，那我们如何来解决它呢？优秀的解决方案往往是 “通过架构来解决问题“，那么能不能通过架构的思想来解决问题呢？

这个架构的核心是通过监听 MySQL 的 binlog 来同步增量数据，通过基于 query 的查询旧表来同步旧数据，这就是本文要讲的一种异构数据库同步的实践。

改进数据库

首先对 MySQL 开启 binlog，但是由于 maxwell 需要的 binlog_format=row 原本的生产环境的数据库不宜修改。这里请教了海杨前辈，他提供了”从库联级“的思路，在从库中监听 binlog 绕过了操作生产环境重启主库的操作，大大降低了系统风险。

这套方案使用到了kafka、maxwell、logstash、elasticsearch。其中 elasticsearch 与 kafka已经在生产环境中有部署，所以无需单独部署维护。而 logstash 与 maxwell 只需要修改配置文件和启动命令即可快速上线。整套方案的意义不仅在于成本低，而且可以大规模使用，公司内有 MySQL 同步到其它数据库的需求时，都可以上任。

成果展示前后对比

方案实施细节

接下来，我们来看看具体是如何实现的。

MySQL

Maxwell（监听 binlog）
Elasticsearch

1. MySQL配置

本次使用 MySQL 5.5 作示范，其他版本的配置可能稍许不同需要

-- 创建一个 用户名为 maxwell 密码为 xxxxxx 的用户
CREATE USER 'maxwell'@'%' IDENTIFIED BY 'XXXXXX';
GRANT ALL ON maxwell.* TO 'maxwell'@'localhost';
GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE ON *.* TO 'maxwell'@'%';

开启数据库的 binlog，修改 mysql 配置文件，注意 maxwell 需要的 binlog 格式必须是row。

# /etc/mysql/my.cnf

[mysqld]
# maxwell 需要的 binlog 格式必须是 row
binlog_format=row

# 指定 server_id 此配置关系到主从同步需要按情况设置，
# 由于此mysql没有开启主从同步，这边默认设置为 1
server_id=1

# logbin 输出的文件名，按需配置
log-bin=master

重启 MySQL 并查看配置是否生效：

sudo systemctl restart mysqld

select @@log_bin;
-- 正确结果是 1
select @@binlog_format;
-- 正确结果是 ROW

如果要监听的数据库开启了主从同步，并且不是主数据库，需要再从数据库开启 binlog 联级同步。

# /etc/my.cnf

log_slave_updates = 1

需要被同步到 elasticsearch 的表结构。

-- robin.logs
show create table robin.logs;

-- 表结构
CREATE TABLE `logs` (
  `id` int(11 unsigned NOT NULL AUTO_INCREMENT,
  `content` text NOT NULL,
  `user_id` int(11 NOT NULL,
  `status` enum('SUCCESS','FAILED','PROCESSING' NOT NULL,
  `type` varchar(20 DEFAULT '',
  `meta` text,
  `created_at` bigint(15 NOT NULL,
  `idx_host` varchar(255 DEFAULT '',
  `idx_domain_id` int(11 unsigned DEFAULT NULL,
  `idx_record_value` varchar(255 DEFAULT '',
  `idx_record_opt` enum('DELETE','ENABLED','DISABLED' DEFAULT NULL,
  `idx_orig_record_value` varchar(255 DEFAULT '',
  PRIMARY KEY (`id`,
  KEY `created_at` (`created_at`
 ENGINE=InnoDB AUTO_INCREMENT=8170697 DEFAULT CHARSET=utf8

2. Maxwell 配置

本次使用 maxwell-1.39.2 作示范, 确保机器中包含 java 环境，推荐 openjdk11

下载 maxwell 程序

wget https://github.com/zendesk/maxwell/releases/download/v1.39.2/maxwell-1.39.2.tar.gz
tar zxvf maxwell-1.39.2.tar.gz **&&**  cd maxwell-1.39.2

maxwell 使用了两个数据库：

另一个是记录maxwell服务状态的数据库，当前这两个数据库可以是同一个

host 需要监听binlog的数据库地址

user 需要监听binlog的数据库用户名
replication_host 记录maxwell服务的数据库地址
replication_user 记录maxwell服务的数据库用户名
producer 将监听到的增量变化数据提交给的消费者 (如 stdout、kafka
kafka_version kafka 版本

启动 maxwell

./bin/maxwell 
        --host=mysql-maxwell.mysql.svc.cluster.fud3 
        --port=3306 
        --user=root 
        --password=password 
        --replication_host=192.168.5.38 
        --replication_port=3306 
        --replication_user=cloner 
        --replication_password=password
        --filter='exclude: *.*, include: robin.logs' 
        --producer=kafka 
        --kafka.bootstrap.servers=192.168.30.10:9092 
        --kafka_topic=maxwell-robinlogs --kafka_version=0.9.0.1

3. 安装 Logstash

Logstash 包中已经包含了 openjdk，无需额外安装。

wget https://artifacts.elastic.co/downloads/logstash/logstash-8.5.0-linux-x86_64.tar.gz
tar zxvf logstash-8.5.0-linux-x86_64.tar.gz

删除不需要的配置文件。

rm config/logstash.yml

修改 logstash 配置文件，此处语法参考官方文档（https://www.elastic.co/guide/en/logstash/current/input-plugins.html）。

# config/logstash-sample.conf

input {
 kafka {
    bootstrap_servers => "192.168.30.10:9092"
    group_id => "main"
    topics => ["maxwell-robinlogs"]
 }
}

filter {
  json {
    source => "message"
  }

  # 将maxwell的事件类型转化为es的事件类型
  # 如增加 -> index 修改-> update
  translate {
    source => "[type]"
    target => "[action]"
    dictionary => {
      "insert" => "index"
      "bootstrap-insert" => "index"
      "update" => "update"
      "delete" => "delete"
    }
    fallback => "unknown"
  }

  # 过滤无效的数据
  if ([action] == "unknown" {
    drop {}
  }

  # 处理数据格式
  if [data][idx_host] {
    mutate {
      add_field => { "idx_host" => "%{[data][idx_host]}" }
    }
  } else {
    mutate {
      add_field => { "idx_host" => "" }
    }
  }

  if [data][idx_domain_id] {
    mutate {
      add_field => { "idx_domain_id" => "%{[data][idx_domain_id]}" }
    }
  } else {
    mutate {
      add_field => { "idx_domain_id" => "" }
    }
  }

  if [data][idx_record_value] {
    mutate {
      add_field => { "idx_record_value" => "%{[data][idx_record_value]}" }
    }
  } else {
    mutate {
      add_field => { "idx_record_value" => "" }
    }
  }
  
   if [data][idx_record_opt] {
    mutate {
      add_field => { "idx_record_opt" => "%{[data][idx_record_opt]}" }
    }
  } else {
    mutate {
      add_field => { "idx_record_opt" => "" }
    }
  }
 
  if [data][idx_orig_record_value] {
    mutate {
      add_field => { "idx_orig_record_value" => "%{[data][idx_orig_record_value]}" }
    }
  } else {
    mutate {
      add_field => { "idx_orig_record_value" => "" }
    }
  }
 
  if [data][type] {
    mutate {
      replace => { "type" => "%{[data][type]}" }
    }
  } else {
    mutate {
      replace => { "type" => "" }
    }
  }
 
  mutate {
    add_field => {
      "id" => "%{[data][id]}"
      "content" => "%{[data][content]}"
      "user_id" => "%{[data][user_id]}"
      "status" => "%{[data][status]}"
      "meta" => "%{[data][meta]}"
      "created_at" => "%{[data][created_at]}"
    }
    remove_field => ["data"]
  }

  mutate {
    convert => {
      "id" => "integer"
      "user_id" => "integer"
      "idx_domain_id" => "integer"
      "created_at" => "integer"
    }
  }

  # 只提炼需要的字段
  mutate {
    remove_field => [
      "message",
      "original",
      "@version",
      "@timestamp",
      "event",
      "database",
      "table",
      "ts",
      "xid",
      "commit",
      "tags"
    ]
   }
}

output {
  # 结果写到es
  elasticsearch {
    hosts => ["http://es-zico2.service.upyun:9500"]
    index => "robin_logs"
    action => "%{action}"
    document_id => "%{id}"
    document_type => "robin_logs"
  }

  # 结果打印到标准输出
  stdout {
    codec => rubydebug
  }
}

执行程序：

# 测试配置文件*
bin/logstash -f config/logstash-sample.conf --config.test_and_exit

# 启动*
bin/logstash -f config/logstash-sample.conf --config.reload.automatic

4. 全量同步

完成启动后，后续的增量数据 maxwell 会自动推送给 logstash 最终推送到 elasticsearch，而之前的旧数据可以通过 maxwell 的 bootstrap 来同步，往下面表中插入一条任务，那么 maxwell 会自动将所有符合条件的 where_clause 的数据推送更新。

INSERT INTO maxwell.bootstrap 
        ( database_name, table_name, where_clause, client_id  
values 
        ( 'robin', 'logs', 'id > 1', 'maxwell' ;

后续可以在 elasticsearch 检测数据是否同步完成，可以先查看数量是否一致，然后抽样对比详细数据。

# 检测 elasticsearch  中的数据量
GET robin_logs/robin_logs/_count

编程笔记 » 如何高效实现 MySQL 与 elasticsearch 的数据同步