mysql 与 es 数据同步常见方案

mysql 与 es 数据同步常见方案

说明

1. @author JellyfishMIX - github / blog.jellyfishmix.com
2. LICENSE GPL-2.0

问题背景

最近需求需要使用 es，并要求 mysql 与 es 做到秒级别的数据同步。梳理了一些 mysql 与 es 间数据同步的常见方案。

同步方案

mysql 与 es 数据同步，目前方案主要有：

双写

写 mysql 的同时写 es。

优点：这种方式简单粗暴，实时写入能做到秒级。

缺点：这种方式代码侵入性强，要在之前写 mysql 的地方加写 es 的代码。以后写 mysql 的地方也要加写 es 的代码。

解析 binlog + kafka

解析 binlog -> 发 kafka 消息 -> 消费 kafka 消息 -> 写 es。

(这里的 kafka 代指消息队列 message queue，具体 mq 选什么可自行替换)

优点：这种方式的好处是代码侵入性低，不需要在之前写 mysql 的地方加写 es 的代码。以后写 mysql 的地方也不需要再额外地写 es。

缺点：这种方式做不到秒级同步，比如kafka 那里消息如果有堆积。即使 kafka 没有消息堆积，这个流程走下来 1 秒也很难完成。而且这种方式的同步延时和数据量有关系。比如一次写十万行，和一次写一两行，数据同步的延时不同。

logstash

logstash 并不局限于收集日志，它本身是一个通道的概念。收集的数据从一端进入，从另一端出来时灌入 es。可以理解为是专门把收集的数据传输至 es 的一个通道。

logstash 可以编写脚本，脚本可以定时地被调度。脚本由同事提供，敏感信息已屏蔽，和真实 sql 不同。由于屏蔽敏感信息，脚本可能不正确，仅供示意，logstash 脚本的编写请另行参考。

input {
    stdin {
    }
    log {
       # log 的位置
    }
    jdbc {
      #jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/my_testdb?characterEncoding=UTF-8&serverTimezone=Asia/Shanghai"
      jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/myTable?characterEncoding=UTF-8&serverTimezone=Asia/Shanghai"
      jdbc_user => "myUser"
      jdbc_password => "myPass"
      jdbc_driver_library => "D:\myES\logstash-7.4.1\logstash-core\lib\jars\mysql-connector-java-5.1.48.jar"
      jdbc_driver_class => "com.mysql.jdbc.Driver"
      statement => "SELECT
                        *
                    FROM
                        extension
                    WHERE
                        type = 1
                    "
      schedule => "*/1 * * * * *"
      # 是否记录上次执行结果，true表示会将上次执行结果的tracking_column字段的值保存到last_run_metadata_path指定的文件中；
      record_last_run => true
      # 需要记录查询结果某字段的值时，此字段为true，否则默认tracking_column为timestamp的值；
      use_column_value => true
      # 需要记录的字段，用于增量同步，需是数据库字段
      #tracking_column => "monitor_time"
      tracking_column => "unix_ts"

      #tracking_column_type => "timestamp"
      # record_last_run上次数据存放位置；
      last_run_metadata_path => "D:\myES\logstash-7.4.1\bin\mysql\last_value\sql_last_value.txt"
      # 是否清除last_run_metadata_path的记录，需要增量同步时此字段必须为false；
      clean_run => false
    }
}
# gsub => ["extension_config", , "/"]
# gsub => [ # 用下划线替换所有正斜杠"fieldname" , "/" , "_" , #用点 "."替换反斜杠、问号、哈希和减号# "fieldname2" , "[\\?#-]" , "." ] } }
filter {
  mutate {
    gsub => ["extension_config","[\\\*]", ""]
  }
}

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

}

上述脚本中 input 内容表示收集数据的来源。

input 中编写了 sql，用来查询从 mysql 中要同步到 es 的数据。input 中后续脚本对数据格式进行了整理。

output 中的内容，表示把收集整理好的数据，灌入 es。

如果每秒调度一次 logstash 的脚本，就可以每秒执行一次 sql 收集 mysql 表中的数据，然后灌入 es。

这种方式能否实现秒级的数据同步，取决于 mysql 表中的数据量和用来收集数据的 sql 执行时间。如果 mysql 表中数据量不大且 sql 能在 1 秒内执行完毕，这种方式实现秒级的数据同步还是有希望的。当然能否真正做到秒级的数据同步，除了 sql 以外，其他环节（比如整理数据格式，传输至 es）的耗时也要考虑。

极端方案 – 同步写 es，异步落库

如果必须用 es，且对于数据实时性、接口响应耗时要求特别高，可以考虑异步落库的方案。实时写 es，异步地发 kafka 消息写数据库。

当然这样异步落库需要关注数据一致性的问题，如果发生了数据不一致，要考虑对 mysql 补偿的机制。

(这里的 kafka 代指消息队列 message queue，具体 mq 选什么可自行替换)