5分鍾搞定 MySQL 到 ClickHouse 實時數據同步進階篇-CloudCanal實戰

本文轉載自查看原文 2022-01-07 09:57 3072

簡述

之前的文章 5分鍾搞定 MySQL 到 ClickHouse 實時數據同步發布后，很多用戶將 MySQL->ClickHouse 實時同步鏈路用了起來，但是我們很快發現，CollapsingMergeTree 在某些場景下可能並不能按預期進行數據折疊。

這個時候，我們參考了 ClickHouse 官方實現的 MaterializeMySQL 表引擎，將 ReplacingMergeTree 作為對端主力表引擎進行數據鏈路構建。

新方案優勢包括

表不存在額外的字段
按 order by key 嚴格合並(單節點)
可以設置按時間間隔自動 optimize 表(autoOptimizeThresholdSec參數)
支持 DDL 同步

技術點

結構遷移

目前到 ClickHouse 的結構遷移中，默認選擇 ReplacingMergeTree 作為表引擎，源主鍵作為 sortKey (無主鍵表則是 tuple)，如下示例:

 CREATE TABLE console.worker_stats
(
    `id` Int64,
    `gmt_create` DateTime,
    `worker_id` Int64,
    `cpu_stat` String,
    `mem_stat` String,
    `disk_stat` String
)
ENGINE = ReplacingMergeTree()
ORDER BY id
SETTINGS index_granularity = 8192

寫數據

新方案全量仍然按照標准 batch 導入，增量和 CollapsingMergeTree 作為表引擎的區別在於

轉換 Insert、Update 操作為 Insert
Delete 操作單獨通過 alter table delete 語句進行操作

所以 Delete 操作如果較多，增量同步性能會急劇下降，建議 delete RPS 不超過 50。

  switch (rowChange.getEventType()) {
            case INSERT:
            case UPDATE: {
                for (CanalRowData rowData : rowChange.getRowDatasList()) {
                    CkTableBatchData.RecordWithState addRecord = new CkTableBatchData.RecordWithState(CanalEventType.INSERT, rowData.getAfterColumnsList());
                    batchData.getRecords().add(addRecord);
                }

                break;
            }
            case DELETE: {
                for (CanalRowData rowData : rowChange.getRowDatasList()) {
                    CkTableBatchData.RecordWithState delRecord = new CkTableBatchData.RecordWithState(CanalEventType.DELETE, rowData.getBeforePkColumnsList());
                    batchData.getRecords().add(delRecord);
                    batchData.setHasDelete(true);
                }
                break;
            }
            default:
                throw new CanalException("not supported event type,eventType:" + rowChange.getEventType());
 }