当前位置：首页 > news >正文

Apache Doris 实战教程：手把手实现 ClickHouse 表结构迁移与数据校验

news 2026/7/24 23:29:03

阅读前提

本教程假设你已具备：

基本的 ClickHouse 使用经验（了解 MergeTree、分布式表概念）
已部署 Apache Doris 集群（单节点或集群均可）
会使用基本的 SQL 和 Shell 命令

第一步：环境准备 & 连接 Doris

首先确认 Doris 集群可连接，并创建目标 Database：

-- 连接 Doris（默认 MySQL 协议，端口 9030） -- mysql -h 127.0.0.1 -P 9030 -u root -- 创建目标数据库 CREATE DATABASE IF NOT EXISTS migration_demo; -- 查看集群状态 SHOW BACKENDS; SHOW FRONTENDS;

第二步：DDL 自动转换——从 CK 到 Doris

2.1 核心映射规则

ClickHouse 和 Apache Doris 在数据类型、表引擎、分区策略上存在差异，需要做以下映射：

# ck_to_doris_ddl.py —— DDL 类型映射核心代码 # ClickHouse → Doris 类型的映射表（覆盖快手支持的 22 种） TYPE_MAP = { "UInt8": "TINYINT", "UInt16": "SMALLINT", "UInt32": "INT", "UInt64": "BIGINT", "Int8": "TINYINT", "Int16": "SMALLINT", "Int32": "INT", "Int64": "BIGINT", "Float32": "FLOAT", "Float64": "DOUBLE", "String": "VARCHAR(65533)", "FixedString": "CHAR", "Date": "DATE", "DateTime": "DATETIME", "DateTime64": "DATETIME", "Array": "ARRAY", "Map": "MAP", "LowCardinality": "VARCHAR", # LowCardinality 展开为 VARCHAR "Enum8": "TINYINT", # Enum 通常转为基础类型 "Enum16": "SMALLINT", } def convert_type(ck_type): """转换 ClickHouse 类型到 Doris 类型""" # 处理 Nullable(Type) if ck_type.startswith("Nullable("): inner = ck_type[9:-1] return convert_type(inner) # Doris 默认支持 NULL，去除 Nullable 包装 # 处理 Array(Type) if ck_type.startswith("Array("): inner = ck_type[6:-1] return f"ARRAY<{convert_type(inner)}>" # 基本类型映射 return TYPE_MAP.get(ck_type, "VARCHAR(65533)") # 未知类型兜底为 VARCHAR

2.2 生成完整的 Doris DDL

# 接上文件 ck_to_doris_ddl.py import re def generate_doris_ddl(ck_table_name, ck_columns, ck_partition_by, ck_order_by): """ 根据 ClickHouse 表信息生成 Doris DDL 参数: ck_table_name: ClickHouse 表名 ck_columns: [(col_name, col_type), ...] ck_partition_by: ClickHouse 分区表达式 ck_order_by: ClickHouse 排序键列名列表 """ doris_table = ck_table_name.replace(".", "_") # 1. 生成列定义 col_defs = [] for col_name, col_type in ck_columns: doris_type = convert_type(col_type) col_defs.append(f" {col_name} {doris_type}") # 2. 从 ORDER BY 推导 DUPLICATE KEY（前两列作为 Key） key_cols = ck_order_by[:2] if len(ck_order_by) >= 2 else ck_order_by # 3. 分区策略：取第一个 Date/DateTime 列做 RANGE 分区 partition_col = ck_partition_by if ck_partition_by else "event_date" # 4. 分桶：按第一列 HASH，默认 32 桶 bucket_col = ck_order_by[0] if ck_order_by else ck_columns[0][0] bucket_num = 32 # 可根据分区数据量调整 ddl = f""" CREATE TABLE IF NOT EXISTS {doris_table} ( {','.join(col_defs)} ) ENGINE = OLAP DUPLICATE KEY({','.join(key_cols)}) PARTITION BY RANGE({partition_col}) () DISTRIBUTED BY HASH({bucket_col}) BUCKETS {bucket_num} PROPERTIES ( "replication_num" = "3" ); """ return ddl.strip() # ---------- 使用示例 ---------- if __name__ == "__main__": # 模拟一个 ClickHouse 表结构 ck_columns = [ ("event_date", "Date"), ("user_id", "UInt64"), ("event_type", "String"), ("event_value", "Float64"), ("event_time", "DateTime"), ("tags", "Array(String)"), ("ext_info", "Map(String, String)"), ] ck_order_by = ["user_id", "event_time"] ddl = generate_doris_ddl( ck_table_name="db.ck_user_behavior", ck_columns=ck_columns, ck_partition_by="event_date", ck_order_by=ck_order_by ) print("-- 生成的 Doris DDL --") print(ddl)

执行结果将生成类似以下 DDL：

-- 生成的 Doris DDL -- CREATE TABLE IF NOT EXISTS db_ck_user_behavior ( event_date DATE, user_id BIGINT, event_type VARCHAR(65533), event_value DOUBLE, event_time DATETIME, tags ARRAY<VARCHAR(65533)>, ext_info MAP<VARCHAR(65533),VARCHAR(65533)> ) ENGINE = OLAP DUPLICATE KEY(user_id, event_time) PARTITION BY RANGE(event_date) () DISTRIBUTED BY HASH(user_id) BUCKETS 32 PROPERTIES ( "replication_num" = "3" );

第三步：配置双跑——数据同时写入 CK 和 Doris

迁移期间，新增数据需要同时写入 ClickHouse 和 Apache Doris：

#!/bin/bash # dual_write.sh —— 离线数据双写脚本 DORIS_HOST="127.0.0.1" DORIS_PORT="8030" # Stream Load HTTP 端口 DORIS_USER="root" DORIS_DB="migration_demo" DORIS_TABLE="user_behavior" # 假设已有 Hive → CK 的导出文件 data_export.csv DATA_FILE="/data/export/data_export.csv" # Stream Load 写入 Doris curl --location-trusted -u ${DORIS_USER}: \ -H "label:dual_write_$(date +%s)" \ -H "column_separator:," \ -H "format:csv" \ -T ${DATA_FILE} \ http://${DORIS_HOST}:${DORIS_PORT}/api/${DORIS_DB}/${DORIS_TABLE}/_stream_load echo "Dual write completed: $(date)"

第四步：存量数据迁移——Hive 到 Doris

对于上游 Hive 数据完整的情况，直接复用已有 ETL 链路：

-- 通过 Doris Multi-Catalog 直接查询 Hive 并 INSERT INTO 目标表 -- 适用于 Hive 数据保留周期够长的场景 -- 1. 创建 Hive Catalog CREATE CATALOG hive_catalog PROPERTIES ( "type" = "hms", "hive.metastore.uris" = "thrift://hive-metastore:9083" ); -- 2. 将 Hive 数据批量写入 Doris 内表 INSERT INTO migration_demo.user_behavior SELECT event_date, user_id, event_type, event_value, event_time FROM hive_catalog.ods.user_behavior_hive WHERE event_date >= '2025-01-01'; -- 3. 查看导入状态 SHOW LOAD FROM migration_demo;

第五步：数据校验——确保迁移结果可靠

5.1 基础数据量校验

-- 第一层校验：行数和基础统计 -- 在 Doris 中执行 SELECT 'Doris' AS source, COUNT(*) AS row_count, SUM(event_value) AS total_value, COUNT(DISTINCT user_id) AS unique_users FROM migration_demo.user_behavior UNION ALL -- 对比 ClickHouse（需单独查询后贴入） -- 预期结果：row_count 和 total_value 应一致 SELECT 'ClickHouse' AS source, 1000000, 12345678.90, 50000;

5.2 维度聚合校验

-- 第二层：按核心维度 GROUP BY 对比 SELECT event_date, event_type, COUNT(*) AS cnt, SUM(event_value) AS sum_val, AVG(event_value) AS avg_val FROM migration_demo.user_behavior WHERE event_date >= '2025-06-01' GROUP BY event_date, event_type ORDER BY event_date, event_type LIMIT 100; -- 将此结果与 ClickHouse 中相同查询结果逐行对比

5.3 Float 精度校验

# float_check.py —— 精度容忍校验 import math def check_float_precision(ck_value, doris_value, tolerance=0.001): """ 检查 Float 类型精度偏差参数: ck_value: ClickHouse 侧数值 doris_value: Doris 侧数值 tolerance: 相对容忍阈值（默认 0.1%） """ if ck_value == 0 and doris_value == 0: return True relative_error = abs(ck_value - doris_value) / max(abs(ck_value), abs(doris_value)) if relative_error > tolerance: print(f"⚠ 精度超标: CK={ck_value}, Doris={doris_value}, error={relative_error:.6f}") return False return True # 使用示例 assert check_float_precision(3.14159265, 3.14159) # True（偏差约 0.00008%） assert not check_float_precision(3.14, 3.25) # False（偏差约 3.5%，超标）

第六步：灰度切换

#!/bin/bash # gray_switch.sh —— 通过 OneSQL 或其他路由层逐步切换 # 1. 10% 流量打向 Doris echo "Gray release: 10% traffic" # onesql-cli set-route --table user_behavior --target doris --weight 10 # 2. 观察 24 小时，确认无异常 # 3. 50% 流量 echo "Gray release: 50% traffic" # onesql-cli set-route --table user_behavior --target doris --weight 50 # 4. 观察 24 小时 # 5. 100% 流量 echo "Full switch to Doris" # onesql-cli set-route --table user_behavior --target doris --weight 100