MySQL到ClickHouse的高速公路-MaterializeMySQL引擎

FavonianKong 发表于 2021/01/20 09:50:00 2021/01/20
【摘要】 MySQL到ClickHouse数据同步原理及实践。

引言

熟悉MySQL的朋友应该都知道,MySQL集群主从间数据同步机制十分完善。令人惊喜的是,ClickHouse作为近年来炙手可热的大数据分析引擎也可以挂载为MySQL的从库,作为MySQL的 "协处理器" 面向OLAP场景提供高效数据分析能力。早先的方案比较直截了当,通过第三方插件将所有MySQL上执行的操作进行转化,然后在ClickHouse端逐一回放达到数据同步。终于在2020年下半年,Yandex 公司在 ClickHouse 社区发布了MaterializeMySQL引擎,支持从MySQL全量及增量实时数据同步。MaterializeMySQL引擎目前支持 MySQL 5.6/5.7/8.0 版本,兼容 Delete/Update 语句,及大部分常用的 DDL 操作。

基础概念

  • MySQL & ClickHouse

    MySQL一般特指完整的MySQL RDBMS,是开源的关系型数据库管理系统,目前属于Oracle公司。MySQL凭借不断完善的功能以及活跃的开源社区,吸引了越来越多的企业和个人用户。

    ClickHouse是由Yandex公司开源的面向OLAP场景的分布式列式数据库。ClickHouse具有实时查询,完整的DBMS及高效数据压缩,支持批量更新及高可用。此外,ClickHouse还较好地兼容SQL语法并拥有开箱即用等诸多优点。

  • Row Store & Column Store

    MySQL存储采用的是Row Store,表中数据按照 Row 为逻辑存储单元在存储介质中连续存储。这种存储方式适合随机的增删改查操作,对于按行查询较为友好。但如果选择查询的目标只涉及一行中少数几个属性,Row 存储方式也不得不将所有行全部遍历再筛选出目标属性,当表属性较多时查询效率通常较低。尽管索引以及缓存等优化方案在 OLTP 场景中能够提升一定的效率,但在面对海量数据背景的 OLAP 场景就显得有些力不从心了。

    ClickHouse 则采用的是 Column Store,表中数据按照Column为逻辑存储单元在存储介质中连续存储。这种存储方式适合采用 SIMD (Single Instruction Multiple Data) 并发处理数据,尤其在表属性较多时查询效率明显提升。列存方式中物理相邻的数据类型通常相同,因此天然适合数据压缩从而达到极致的数据压缩比。

    Snipaste_2021-01-13_16-11-12.png

使用方法

  • 部署Master-MySQL
    开启BinLog功能:ROW模式
    开启GTID模式:解决位点同步时MySQL主从切换问题(BinLog reset导致位点失效)

    # my.cnf关键配置
    gtid_mode=ON
    enforce_gtid_consistency=1
    binlog_format=ROW
  • 部署Slave-ClickHouse
    获取 ClickHouse/Master 代码编译安装
    推荐使用GCC-10.2.0,CMake 3.15,ninja1.9.0及以上
  • 创建Master-MySQL中database及table
    creat databases master_db;
    use master_db;
    CREATE TABLE IF NOT EXISTS `runoob_tbl`(
       `runoob_id` INT UNSIGNED AUTO_INCREMENT,
       `runoob_` VARCHAR(100) NOT NULL,
       `runoob_author` VARCHAR(40) NOT NULL,
       `submission_date` DATE,
       PRIMARY KEY ( `runoob_id` )
    )ENGINE=InnoDB DEFAULT CHARSET=utf8;
    
    # 插入几条数据
    INSERT INTO runoob_tbl (runoob_, runoob_author, submission_date) VALUES ("MySQL-learning", "Bob", NOW());
    INSERT INTO runoob_tbl (runoob_, runoob_author, submission_date) VALUES ("MySQL-learning", "Tim", NOW());
  • 创建 Slave-ClickHouse 中 MaterializeMySQL database
    # 开启materialize同步功能
    SET allow_experimental_database_materialize_mysql=1;
    # 创建slave库,参数分别是("mysqld服务地址", "待同步库名", "授权账户", "密码")
    CREATE DATABASE slave_db ENGINE = MaterializeMySQL('192.168.6.39:3306', 'master_db', 'root', '3306123456');
    此时可以看到ClickHouse中已经有从MySQL中同步的数据了:
    DESKTOP:) select * from  runoob_tbl;
    
    SELECT *
    FROM runoob_tbl
    
    Query id: 6e2b5f3b-0910-4d29-9192-1b985484d7e3
    
    ┌─runoob_id─┬─runoob_title───┬─runoob_author─┬─submission_date─┐
    │         1 │ MySQL-learning │ Bob           │      2021-01-06 │
    └───────────┴────────────────┴───────────────┴─────────────────┘
    ┌─runoob_id─┬─runoob_title───┬─runoob_author─┬─submission_date─┐
    │         2 │ MySQL-learning │ Tim           │      2021-01-06 │
    └───────────┴────────────────┴───────────────┴─────────────────┘
    2 rows in set. Elapsed: 0.056 sec.

工作原理

  • BinLog Event

    MySQL中BinLog Event主要包含以下几类:

    1. MYSQL_QUERY_EVENT    -- DDL
    2. MYSQL_WRITE_ROWS_EVENT -- insert
    3. MYSQL_UPDATE_ROWS_EVENT -- update
    4. MYSQL_DELETE_ROWS_EVENT -- delete

    事务提交后,MySQL 将执行过的 SQL 处理 BinLog Event,并持久化到 BinLog 文件

    ClickHouse通过消费BinLog达到数据同步,过程中主要考虑3个方面问题:

    1、DDL兼容:由于ClickHouse和MySQL的数据类型定义有区别,DDL语句需要做相应转换

    2、Delete/Update 支持:引入_version字段,控制版本信息

    3、Query 过滤:引入_sign字段,标记数据有效性

  • DDL操作

    对比一下MySQL的DDL语句以及在ClickHouse端执行的DDL语句:

    mysql> show create table runoob_tbl\G;
    *************************** 1. row ***************************
    Table: runoob_tbl
    Create Table: CREATE TABLE `runoob_tbl` (
      `runoob_id` int unsigned NOT NULL AUTO_INCREMENT,
      `runoob_` varchar(100) NOT NULL,
      `runoob_author` varchar(40) NOT NULL,
      `submission_date` date DEFAULT NULL,
      PRIMARY KEY (`runoob_id`)
    ) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8
    1 row in set (0.00 sec)
    ---------------------------------------------------------------
    cat /metadata/slave_db/runoob_tbl.sql
    ATTACH TABLE _ UUID '14dbff59-930e-4aa8-9f20-ccfddaf78077'
    (
        `runoob_id` UInt32,
        `runoob_` String,
        `runoob_author` String,
        `submission_date` Nullable(Date),
        `_sign` Int8 MATERIALIZED 1,
        `_version` UInt64 MATERIALIZED 1
    )
    ENGINE = ReplacingMergeTree(_version)
    PARTITION BY intDiv(runoob_id, 4294967)
    ORDER BY tuple(runoob_id)
    SETTINGS index_granularity = 8192

    可以看到:

    1、在DDL转化时默认增加了2个隐藏字段:_sign(-1删除, 1写入) 和 _version(数据版本)
    2、默认将表引擎设置为 ReplacingMergeTree,以 _version 作为 column version
    3、原DDL主键字段 runoob_id 作为ClickHouse排序键和分区键

    此外还有许多DDL处理,比如增加列、索引等,相应代码在Parsers/MySQL 目录下。

  • Delete/Update操作

    Update:

    # Mysql端:
    UPDATE runoob_tbl set runoob_author='Mike' where runoob_id=2;
    
    mysql> select * from runoob_tbl;
    +-----------+----------------+---------------+-----------------+
    | runoob_id | runoob_title   | runoob_author | submission_date |
    +-----------+----------------+---------------+-----------------+
    |         1 | MySQL-learning | Bob           | 2021-01-06      |
    |         2 | MySQL-learning | Mike          | 2021-01-06      |
    +-----------+----------------+---------------+-----------------+
    2 rows in set (0.00 sec)
    
    ----------------------------------------------------------------
    # ClickHouse端:
    DESKTOP:) select *, _sign, _version from runoob_tbl order by runoob_id;
    
    SELECT
        *,
        _sign,
        _version
    FROM runoob_tbl
    ORDER BY runoob_id ASC
    
    Query id: c5f4db0a-eff6-4b49-a429-b55230c26301
    
    ┌─runoob_id─┬─runoob_title───┬─runoob_author─┬─submission_date─┬─_sign─┬─_version─┐
    │         1 │ MySQL-learning │ Bob           │      2021-01-06 │     1 │        2 │
    │         2 │ MySQL-learning │ Mike          │      2021-01-06 │     1 │        4 │
    │         2 │ MySQL-learning │ Tim           │      2021-01-06 │     1 │        3 │
    └───────────┴────────────────┴───────────────┴─────────────────┴───────┴──────────┘
    3 rows in set. Elapsed: 0.003 sec.
    可以看到,ClickHouse数据也实时同步了更新操作。
  • Delete:

    # Mysql端
    mysql> DELETE from runoob_tbl where runoob_id=2;
    
    mysql> select * from runoob_tbl;
    +-----------+----------------+---------------+-----------------+
    | runoob_id | runoob_title   | runoob_author | submission_date |
    +-----------+----------------+---------------+-----------------+
    |         1 | MySQL-learning | Bob           | 2021-01-06      |
    +-----------+----------------+---------------+-----------------+
    1 row in set (0.00 sec)
    
    ----------------------------------------------------------------
    # ClickHouse端
    DESKTOP:) select *, _sign, _version from runoob_tbl order by runoob_id;
    
    SELECT
        *,
        _sign,
        _version
    FROM runoob_tbl
    ORDER BY runoob_id ASC
    
    Query id: e9cb0574-fcd5-4336-afa3-05f0eb035d97
    
    ┌─runoob_id─┬─runoob_title───┬─runoob_author─┬─submission_date─┬─_sign─┬─_version─┐
    │         1 │ MySQL-learning │ Bob           │      2021-01-06 │     1 │        2 │
    └───────────┴────────────────┴───────────────┴─────────────────┴───────┴──────────┘
    ┌─runoob_id─┬─runoob_title───┬─runoob_author─┬─submission_date─┬─_sign─┬─_version─┐
    │         2 │ MySQL-learning │ Mike          │      2021-01-06 │    -1 │        5 │
    └───────────┴────────────────┴───────────────┴─────────────────┴───────┴──────────┘
    ┌─runoob_id─┬─runoob_title───┬─runoob_author─┬─submission_date─┬─_sign─┬─_version─┐
    │         2 │ MySQL-learning │ Mike          │      2021-01-06 │     1 │        4 │
    │         2 │ MySQL-learning │ Tim           │      2021-01-06 │     1 │        3 │
    └───────────┴────────────────┴───────────────┴─────────────────┴───────┴──────────┘
    4 rows in set. Elapsed: 0.002 sec.

    可以看到,删除id为2的行只是额外插入了_sign == -1的一行记录,并没有真正删掉。

  • 日志回放

    MySQL 主从间数据同步时Slave节点将 BinLog Event 转换成相应的SQL语句,Slave 模拟 Master 写入。类似地,传统第三方插件沿用了MySQL主从模式的BinLog消费方案,即将 Event 解析后转换成 ClickHouse 兼容的 SQL 语句,然后在 ClickHouse 上执行(回放),但整个执行链路较长,通常性能损耗较大。不同的是,MaterializeMySQL 引擎提供的内部数据解析以及回写方案隐去了三方插件的复杂链路。回放时将 BinLog Event 转换成底层 Block 结构,然后直接写入底层存储引擎,接近于物理复制。此方案可以类比于将 BinLog Event 直接回放到 InnoDB 的 Page 中。

同步策略

    • 日志回放

      v20.9.1版本前是基于位点同步的,ClickHouse每消费完一批 BinLog Event,就会记录 Event 的位点信息到 .metadata 文件:

      [FavonianKong@Wsl[20:42:37]slave_db]
      $ cat ./.metadata
      Version:        2
      Binlog File:    mysql-bin.000003
      Binlog Position:355005999
      Data Version:   5
      这样当 ClickHouse 再次启动时,它会把 {‘mysql-bin.000003’, 355005999} 二元组通过协议告知 MySQL Server,MySQL 从这个位点开始发送数据:

      s1> ClickHouse 发送 {‘mysql-bin.000003’, 355005999} 位点信息给 MySQL
      s2> MySQL 找到本地 mysql-bin.000003 文件并定位到 355005999 偏移位置,读取下一个 Event 发送给 ClickHouse
      s3> ClickHouse 接收 binlog event 并完成同步操作
      s4> ClickHouse 更新 .metadata位点

      存在问题:

      如果MySQL Server是一个集群,通过VIP对外服务,MaterializeMySQL创建 database 时 host 指向的是VIP,当集群主从发生切换后,{Binlog File, Binlog Position} 二元组不一定是准确的,因为BinLog可以做reset操作。

      s1> ClickHouse 发送 {'mysql-bin.000003’, 355005999} 给集群新主 MySQL
      s2> 新主 MySQL 发现本地没有 mysql-bin.000003 文件,因为它做过 reset master 操作,binlog 文件是 mysql-bin.000001
      s3> 产生错误复制

      为了解决这个问题,v20.9.1版本后上线了 GTID 同步模式,废弃了不安全的位点同步模式。

    • GTID同步

      GTID模式为每个 event 分配一个全局唯一ID和序号,直接告知 MySQL 这个 GTID 即可,于是.metadata变为:

      [FavonianKong@Wsl[21:30:19]slave_db]
      Version:        2
      Binlog File:    mysql-bin.000003
      Executed GTID:  0857c24e-4755-11eb-888c-00155dfbdec7:1-783
      Binlog Position:355005999
      Data Version:   5

      其中 0857c24e-4755-11eb-888c-00155dfbdec7 是生成 Event的主机UUID,1-783是已经同步的event区间

      于是流程变为:

      s1> ClickHouse 发送 GTID:0857c24e-4755-11eb-888c-00155dfbdec7:1-783 给 MySQL
      s2> MySQL 根据 GTID 找到本地位点,读取下一个 Event 发送给 ClickHouse
      s3> ClickHouse 接收 BinLog Event 并完成同步操作
      s4> ClickHouse 更新 .metadata GTID信息

    源码分析

    • 概述

      在最新源码 (v20.13.1.1) 中,ClickHouse 官方对 DatabaseMaterializeMySQL 引擎的相关源码进行了重构,并适配了 GTID 同步模式。ClickHouse 整个项目的入口 main 函数在 /ClickHouse/programs/main.cpp 文件中,主程序会根据接收指令将任务分发到 ClickHouse/programs 目录下的子程序中处理。本次分析主要关注 Server 端 MaterializeMySQL 引擎的工作流程。

    • 源码目录

      与 MaterializeMySQL 相关的主要源码路径:

      ClickHouse/src/databases/MySQL   //MaterializeMySQL存储引擎实现
      ClickHouse/src/Storages/         //表引擎实现
      ClickHouse/src/core/MySQL*       //复制相关代码
      ClickHouse/src/Interpreters/     //Interpreters实现,SQL的rewrite也在这里处理
      ClickHouse/src/Parsers/MySQL     //解析部分实现,DDL解析等相关处理在这里
    • 服务端主要流程

      ClickHouse 使用 POCO 网络库处理网络请求,Client连接的处理逻辑在 ClickHouse/src/Server/*Handler.cpp 的 hander方法里。以TCP为例,除去握手,初始化上下文以及异常处理等相关代码,主要逻辑可以抽象成:

      // ClickHouse/src/Server/TCPHandler.cpp
      TCPHandler.runImpl()
      {
          ...
          while(true) {
              ...
              if (!receivePacket())  //line 184
                      continue
              /// Processing Query   //line 260
              state.io = executeQuery(state.query, *query_context, ...);
          ...
      }
    • 数据同步预处理

      Client发送的SQL在executeQuery函数处理,主要逻辑简化如下:

      // ClickHouse/src/Interpreters/executeQuery.cpp
      static std::tuple executeQueryImpl(...)
      {
          ...
          // line 354,解析器可配置
          ast = parseQuery(...); 
          ...
          // line 503, 根据语法树生成interpreter
          auto interpreter = InterpreterFactory::get(ast, context, ...);
          ...
          // line 525, 执行器interpreter执行后返回结果
          res = interpreter->execute();
          ...
      }

      主要有三点:

      1、解析SQL语句并生成语法树 AST
      2、InterpreterFactory 工厂类根据 AST 生成执行器
      3、interpreter->execute()

      跟进第三点,看看 InterpreterCreateQuery 的 excute() 做了什么:

      // ClickHouse/src/Interpreters/InterpreterCreateQuery.cpp
      BlockIO InterpreterCreateQuery::execute()
      {
          ...
          // CREATE | ATTACH DATABASE
          if (!create.database.empty() && create.table.empty())
              // line 1133, 当使用MaterializeMySQL时,会走到这里建库
              return createDatabase(create);  
      }

      这里注释很明显,主要执行 CREATE 或 ATTACH DATABASE,继续跟进 createDatabase() 函数:

      // ClickHouse/src/Interpreters/InterpreterCreateQuery.cpp
      BlockIO InterpreterCreateQuery::createDatabase(ASTCreateQuery & create)
      {
          ...
          // line 208, 这里会根据 ASTCreateQuery 参数,从 DatabaseFactory 工厂获取数据库对象
          // 具体可以参考 DatabasePtr DatabaseFactory::getImpl() 函数
          DatabasePtr database = DatabaseFactory::get(create, metadata_path, ...); 
          ...
          // line 253, 多态调用,在使用MaterializeMySQL时
          // 上方get函数返回的是 DatabaseMaterializeMySQL
          database->loadStoredObjects(context, ...);
      }

      到这里,相当于将任务分发给DatabaseMaterializeMySQL处理,接着跟踪 loadStoredObjects 函数:

      //ClickHouse/src/Databases/MySQL/DatabaseMaterializeMySQL.cpp
      template
      void DatabaseMaterializeMySQL::loadStoredObjects(Context & context, ...)
      {
          Base::loadStoredObjects(context, has_force_restore_data_flag, force_attach);
          try
          {
              // line87, 这里启动了materialize的同步线程
              materialize_thread.startSynchronization(); 
              started_up = true;
          }
          catch (...)
        ...
      }

      跟进startSynchronization() 绑定的执行函数:

      // ClickHouse/src/Databases/MySQL/MaterializeMySQLSyncThread.cpp
      void MaterializeMySQLSyncThread::synchronization()
      {
          ...
          // 全量同步在 repareSynchronized() 进行
          if (std::optional metadata = prepareSynchronized())
          {
              while (!isCancelled())
              {
                  UInt64 max_flush_time = settings->max_flush_data_time;
                  BinlogEventPtr binlog_event = client.readOneBinlogEvent(...);
                  {
                      //增量同步侦听binlog_envent
                      if (binlog_event)
                          onEvent(buffers, binlog_event, *metadata);
                  }
              }
          }
        ...
      }

    • 全量同步

      MaterializeMySQLSyncThread::prepareSynchronized 负责DDL和全量同步,主要流程简化如下:

      // ClickHouse/src/Databases/MySQL/MaterializeMySQLSyncThread.cpp
      std::optional MaterializeMySQLSyncThread::prepareSynchronized()
      {
          while (!isCancelled())
          {
              ...
              try
              {
                  //构造函数内会获取MySQL的状态、MySQL端的建表语句,
                  MaterializeMetadata metadata(connection, ...);
                  // line345, DDL相关转换
                  metadata.transaction(position, [&]() 
                  {
                   cleanOutdatedTables(database_name, global_context);
                      dumpDataForTables(connection, metadata, global_context, ...);
                  });
                  
                  return metadata;
              }
              ...
         }
      }

      ClickHouse作为MySQL从节点,在MaterializeMetadata构造函数中对MySQL端进行了一系列预处理:

      1、将打开的表关闭,同时对表加上读锁并启动事务
      2、TablesCreateQuery通过SHOW CREATE TABLE 语句获取MySQL端的建表语句
      3、获取到建表语句后释放表锁

      继续往下走,执行到 metadata.transaction() 函数,该调用传入了匿名函数作为参数,一直跟进该函数会发现最终会执行匿名函数,也就是cleanOutdatedTables以及dumpDataForTables函数,主要看一下 dumpDataForTables 函数:

      // ClickHouse/src/Databases/MySQL/MaterializeMySQLSyncThread.cpp
      static inline void dumpDataForTables(...)
      {
          ...
          //line293, 这里执行建表语句
          tryToExecuteQuery(..., query_context, database_name, comment); 
      }
      继续跟踪 tryToExecuteQuery 函数,会调用到 executeQueryImpl() 函数,上文提到过这个函数,但这次我们的上下文信息变了,生成的执行器发生变化,此时会进行 DDL 转化以及 dump table 等操作:
      // ClickHouse/src/Interpreters/executeQuery.cpp
      static std::tuple executeQueryImpl(...)
      {
          ... 
          // line 354,解析器可配置
          ast = parseQuery(...); 
          ...
          // line 503,这里跟之前上下文信息不同,生成interpreter也不同
          auto interpreter = InterpreterFactory::get(ast,context, ...);
          ...
          // line 525, 执行器interpreter执行后返回结果
          res = interpreter->execute(); 
          ...
      }
      此时 InterpreterFactory 返回 InterpreterExternalDDLQuery,跟进去看 execute 函数做了什么:
      // ClickHouse/src/Interpreters/InterpreterExternalDDLQuery.cpp
      BlockIO InterpreterExternalDDLQuery::execute()
      {
          ...
          if (external_ddl_query.from->name == "MySQL")
          {
      #ifdef USE_MYSQL
              ...
              // line61, 当全量复制执行DDL时,会执行到这里
              else if (...->as())
                  return MySQLInterpreter::InterpreterMySQLCreateQuery(
                  external_ddl_query.external_ddl, cogetIdentifierName(arguments[0]),
                  getIdentifierName(arguments[1])).execute();
      #endif
          }
       ...
        return BlockIO();
      }
      继续跟进去看看 getIdentifierName(arguments[1])).execute() 做了什么事情:
      // ClickHouse/src/Interpreters/MySQL/InterpretersMySQLDDLQuery.h
      class InterpreterMySQLDDLQuery : public IInterpreter
      {
          public:
          ...
          BlockIO execute() override
          {
              ...
              // line68, 把从MySQL获取到的DDL语句进行转化
              ASTs rewritten_queries = InterpreterImpl::getRewrittenQueries(
                         query, context, mapped_to_database, mysql_database);
              
              // line70, 这里执行转化后的DDL语句
              for (const auto & rewritten_query : rewritten_queries)
                  executeQuery(..., queryToString(rewritten_query), ...);
      
              return BlockIO{};
          }
          ...
      }
      进一步看 InterpreterImpl::getRewrittenQueries 是怎么转化 DDL 的:
      // ClickHouse/src/Interpreters/MySQL/InterpretersMySQLDDLQuery.cpp
      ASTs InterpreterCreateImpl::getRewrittenQueries(...)
      {
          ...
          // 检查是否存在primary_key, 没有直接报错
          if (primary_keys.empty())
              throw Exception("cannot be materialized, no primary keys.", ...);
          ...
          // 添加 _sign 和 _version 列.
          auto sign_column_name = getUniqueColumnName(columns_name_and_type, "_sign");
          auto version_column_name = getUniqueColumnName(columns_name_and_type, "_version");
      
          // 这里悄悄把建表引擎修改成了ReplacingMergeTree
          storage->set(storage->engine, makeASTFunction("ReplacingMergeTree", ...));
          ...
          return ASTs{rewritten_query};
      }
      完成DDL转换之后就会去执行新的DDL语句,完成建表操作,再回到 dumpDataForTables:
      // ClickHouse/src/Databases/MySQL/MaterializeMySQLSyncThread.cpp
      static inline void dumpDataForTables(...)
      {
          ...
          //line293, 这里执行建表语句
          tryToExecuteQuery(..., query_context, database_name, comment);
          ...
          // line29, 这里开始 dump 数据并存放到MySQLBlockInputStream
         MySQLBlockInputStream input(connection, ...);
      }
    • 增量同步

      还记得startSynchronization() 绑定的执行函数吗?全量同步分析都是在 prepareSynchronized()进行的,那增量更新呢?

      // ClickHouse/src/Databases/MySQL/MaterializeMySQLSyncThread.cpp
      void MaterializeMySQLSyncThread::synchronization()
      {
          ...
          // 全量同步在 repareSynchronized() 进行
          if (std::optional metadata = prepareSynchronized())
          {
              while (!isCancelled())
              {
                  UInt64 max_flush_time = settings->max_flush_data_time;
                  BinlogEventPtr binlog_event = client.readOneBinlogEvent(...);
                  {
                      //增量同步侦听binlog_envent
                      if (binlog_event)
                          onEvent(buffers, binlog_event, *metadata);
                  }
              }
          }
          ...
      }

      可以看到,while 语句里有一个 binlog_event 的侦听函数,用来侦听 MySQL 端 BinLog 日志变化,一旦 MySQL 端执行相关操作,其 BinLog 日志会更新并触发 binlog_event,增量更新主要在这里进行。

      // ClickHouse/src/Databases/MySQL/MaterializeMySQLSyncThread.cpp
      void MaterializeMySQLSyncThread::onEvent(Buffers & buffers, const BinlogEventPtr & receive_event, MaterializeMetadata & metadata)
      { 
          // 增量同步通过监听binlog event实现,目前支持四种event:MYSQL_WRITE_ROWS_EVENT、
          // MYSQL_UPDATE_ROWS_EVENT、MYSQL_DELETE_ROWS_EVENT 和 MYSQL_QUERY_EVENT
          // 具体的流程可以查找对应的 onHandle 函数, 不在此详细分析
          if (receive_event->type() == MYSQL_WRITE_ROWS_EVENT){...}
          else if (receive_event->type() == MYSQL_UPDATE_ROWS_EVENT){...}
          else if (receive_event->type() == MYSQL_DELETE_ROWS_EVENT){...}
          else if (receive_event->type() == MYSQL_QUERY_EVENT){...}
          else {/* MYSQL_UNHANDLED_EVENT*/}
      }

    小结

    MaterializeMySQL 引擎是 ClickHouse 官方2020年主推的特性,由于该特性在生产环境中属于刚需且目前刚上线不久,整个模块处于高速迭代的状态,因此有许多待完善的功能。例如复制过程状态查看以及数据的一致性校验等。感兴趣的话可参考Github上的2021-Roadmap,里面会更新一些社区最近得计划。以上内容如有理解错误还请指正。

    引用

    ClickHouse社区文档

    ClickHouse社区源码

    MySQL实时复制与实现

    MaterializeMySQL引擎分析

    【版权声明】本文为华为云社区用户原创内容,转载时必须标注文章的来源(华为云社区),文章链接,文章作者等基本信息,否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件至:cloudbbs@huaweicloud.com进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容。
    • 点赞
    • 收藏
    • 关注作者

    评论(0

    0/1000
    抱歉,系统识别当前为高风险访问,暂不支持该操作

    全部回复

    上滑加载中

    设置昵称

    在此一键设置昵称,即可参与社区互动!

    *长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

    *长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。