同步方式

主要功能

适用场景

CTAS

分库分表合并同步

适用互动业务多数场景（方便维护）

CDAS

整库同步、多表同步

数据验证

启动方式

验证方式

验证结果

无状态启动-全量

1. 数据库插入 1000 条数据
2. 启动 Flink 采集任务
3. 观察 Holo 目标表数据

目标表1000条数据

无状态启动-指定点位

（只有SLS/kafka可用）

1. 创建 Flink任务采集SLS日志至Holo
2. 启动时设置点位为 2023-10-18 13:00:00
3. 查询Holo表最早的一条数据时间

结果：最早的一条数据时间为

2023-10-18 12:59:59

从最新状态恢复-增量

1. 直接启动任务

结果：任务启动时刻开始数据采集

从指定状态恢复-快照

1. 启动 Flink 采集任务
2. 数据库插入 1000 条数据
3. 创建快照，停止 Flink 任务
4. 数据库Insert Into 2000 条数据
5. 开启 Flink 任务

结果：FLink同步了第4点中插入的2000条数据，并且自增id是连续的。

源端

目标端

验证验证

MySQL

Holo

-- 定义数据源表
CREATE TEMPORARY TABLE user_clicks_source
(db_name         STRING METADATA FROM 'database_name' VIRTUAL  -- 读取库名。,table_name     STRING METADATA FROM 'table_name' VIRTUAL  -- 读取表名。,operation_ts   TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL  -- 读取变更时间。,username       STRING,click_url      STRING,eventtime      STRING,operation_type STRING
)
with ('connector' = 'mysql','hostname' = 'rm-2zer8jdp1xjg6bh3upo.mysql.rds.aliyuncs.com','port' = '3306','username' = 'xxx','password' = 'xxx','database-name' = 'double_v_test','table-name' = 'user_clicks','scan.incremental.snapshot.chunk.key-column' = 'id'
)
;-- 定义结果表
CREATE TEMPORARY TABLE user_clicks_sink
(db_name         STRING,table_name     STRING,operation_ts   TIMESTAMP,username       STRING,click_url      STRING,eventtime      STRING,operation_type STRING
)
WITH ('connector' = 'hologres','dbname' = 'ods','tablename' = 'user_clicks','username' = 'xxx','password' = 'xxx','endpoint' = 'hgpostcn-cn-g4t3fc54t007-cn-beijing-vpc-st.hologres.aliyuncs.com:80'
)
;
-- 写入数据到结果表
INSERT INTO user_clicks_sink
SELECTdb_name,table_name,operation_ts,username,click_url,eventtime,operation_type
FROM user_clicks_source
;

SLS

Holo

-- 定于源表
CREATE TEMPORARY TABLE sls_source(content VARCHAR
) WITH ('connector' = 'sls','endPoint' = 'cn-beijing.log.aliyuncs.com','project' = 'bigdata-spm-prod','logStore' = 'bigdata-spm','accessId' = 'LTAI4GJpG4GiVLjDETaxiAWJ','accessKey' = 'O5OfRSXLuaFvbZ3mEFNyteF6ukofVR'
);-- 定义结果表
CREATE TEMPORARY TABLE sls_spm (content VARCHAR
) WITH ('connector' = 'hologres','dbname' = 'ods','tablename' = 'sls_spm','username' = 'LTAI5t9BrCrWCgMnLR7cpW4K','password' = 'oIDPmdegfy3j9Z371kPBPEyJyMDHPY','endpoint' = 'hgpostcn-cn-g4t3fc54t007-cn-beijing-vpc-st.hologres.aliyuncs.com:80'
);-- Insert Into 语句
INSERT INTO sls_spm 
SELECT content FROM sls_source
;

Holo

MySQL

-- 定义数据源表
CREATE TEMPORARY TABLE user_clicks_source
(username   STRING,click_url STRING,eventtime STRING
)
WITH ('connector' = 'hologres','dbname' = 'ods','tablename' = 'user_clicks','username' = 'xxx','password' = 'xxx','endpoint' = 'hgpostcn-cn-g4t3fc54t007-cn-beijing-vpc-st.hologres.aliyuncs.com:80','binlog' = 'true'  -- 开启binlog,'cdcMode' = 'true'  -- 使用CDC模式,'binlogMaxRetryTimes' = '10'  -- 读取 Binlog 时最大重试次数,'binlogRetryIntervalMs' = '500'  -- 重试的时间间隔，这里设置为 500ms,'binlogBatchReadSize' = '100'  -- 每批读取 Binlog 的大小，这里设置为100条记录
)
;-- 定义结果表
CREATE TEMPORARY TABLE user_clicks_sink
(username   STRING,click_url STRING,eventtime STRING
)
with ('connector' = 'mysql','hostname' = 'rm-2zer8jdp1xjg6bh3upo.mysql.rds.aliyuncs.com','port' = '3306','username' = 'xxx','password' = 'xxx','database-name' = 'double_v_test','table-name' = 'user_clicks'
)
;-- Insert Into 语句
INSERT INTO user_clicks_sink
SELECTusername,click_url,eventtime
FROM user_clicks_source
;

注：需建表时开启Binlog

Mysql

MaxComputer

-- 定义数据源表
CREATE TEMPORARY TABLE user_clicks_source
(db_name         STRING METADATA FROM 'database_name' VIRTUAL  -- 读取库名。,table_name     STRING METADATA FROM 'table_name' VIRTUAL  -- 读取表名。,operation_ts   TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL  -- 读取变更时间。,username       STRING,click_url      STRING,eventtime      STRING,operation_type STRING
)
with ('connector' = 'mysql','hostname' = 'rm-2zer8jdp1xjg6bh3upo.mysql.rds.aliyuncs.com','port' = '3306','username' = 'xxx','password' = 'xxx','database-name' = 'double_v_test','table-name' = 'user_clicks','scan.incremental.snapshot.chunk.key-column' = 'id'
)
;-- 定义结果表
CREATE TEMPORARY TABLE user_clicks_sink
(db_name         STRING,table_name     STRING,operation_ts   STRING,username       STRING,click_url      STRING,eventtime      STRING,operation_type STRING
)
WITH ('connector' = 'odps','endpoint' = 'http://service.cn-beijing.maxcompute.aliyun.com/api','project' = 'bigdata_spm_produce','tablename' = 'user_clicks','accessid' = 'xxx','accessKey' = 'xxx'
)
;-- Insert Into 语句
INSERT INTO user_clicks_sink
SELECTdb_name,table_name,CAST(operation_ts AS STRING),username,click_url,eventtime,operation_type
FROM user_clicks_source
;

Holo

MaxComputer

-- 定义数据源表
CREATE TEMPORARY TABLE user_clicks_source
(username   STRING,click_url STRING,eventtime STRING
)
WITH ('connector' = 'hologres','dbname' = 'ods','tablename' = 'user_clicks','username' = 'xxx','password' = 'xxx','endpoint' = 'hgpostcn-cn-g4t3fc54t007-cn-beijing-vpc-st.hologres.aliyuncs.com:80','binlog' = 'true'  -- 开启binlog,'cdcMode' = 'true'  -- 使用CDC模式,'binlogMaxRetryTimes' = '10'  -- 读取 Binlog 时最大重试次数,'binlogRetryIntervalMs' = '500'  -- 重试的时间间隔，这里设置为 500ms,'binlogBatchReadSize' = '100'  -- 每批读取 Binlog 的大小，这里设置为100条记录
)
;-- 定义结果表
CREATE TEMPORARY TABLE user_clicks_sink
(username       STRING,click_url      STRING,eventtime      STRING
)
WITH ('connector' = 'odps','endpoint' = 'http://service.cn-beijing.maxcompute.aliyun.com/api','project' = 'bigdata_spm_produce','tablename' = 'user_clicks','accessid' = 'xxx','accessKey' = 'xxx'
)
;-- Insert Into 语句
INSERT INTO user_clicks_sink
SELECTusername,click_url,eventtime
FROM user_clicks_source
;

SLS

MaxComputer

-- 定义源表CREATE TEMPORARY TABLE sls_spm_source
(content VARCHAR
)
WITH ('connector' = 'sls','endPoint' = 'cn-beijing.log.aliyuncs.com','project' = 'bigdata-spm-prod','logStore' = 'bigdata-spm','accessId' = 'xxx','accessKey' = 'xxx'
)
;--定义结果表CREATE TEMPORARY TABLE sls_spm_sink
(content STRING
)
WITH ('connector' = 'odps','endpoint' = 'http://service.cn-beijing.maxcompute.aliyun.com/api','project' = 'bigdata_spm_produce','tablename' = 'sls_spm','accessid' = 'xxx','accessKey' = 'xxx'
)
;
-- Insert Into 语句
INSERT INTO sls_spm_sink
SELECTcontent
FROM sls_spm_source
;

类型

功能

同步

全增量一体化：全量和增量数据一体化读取，全增量自动切换，无需维护两条链路

并发读取：全量阶段支持多并发读取，水平扩展读取性能

无锁读取：基于无锁一致性算法，无需数据库加锁，不影响在线业务

断点续传：全量和增量阶段均支持Checkpoint，作业快速恢复，稳定性高

ETL：无缝集成Flink SQL，支持对数据库数据做灵活强大的 Streaming ETL加工

轻量化：无需部署额外的 Kafka、Canal 等服务，链路短、成本低、易维护

分库分表合并：支持分库分表合并入仓入湖

表结构变更同步：源表schema变更（加列等）自动同步到目标表

整库同步：支持整库读取，一行 SQL 完成整库同步作业开发

千表入仓：单作业支撑上千张规模的不同业务表实时写入数据仓库

元数据：内置统一元敛据管理，自动发现外部元数据（MySQL等），无需手写DDL

数据限流：提供限定同步速率的能力，避免给源端数据库造成太大压力，影响在线业务。

脏数据收集：脏数据自动收集到 OSS/Kafka/SLS，不影响同步链路稳定性，提供脏数据事后处理能力。

数据质量校验：提供数据同步的目标数据与原始数据的一致性校验服务。

指标

含义

职责

场景

并发度

控制作业并行程度（并行执行的任务数量）

它控制了作业的并行程度，决定了作业在集群中的资源使用情况。

适用于需要高吞吐量和低延迟的场景。增加并发度可以提高作业的整体性能。

JobManager CPU

管理作业的整体协调和调度

JobManager负责协调作业的执行、状态管理和故障恢复。

管理作业和调度的计算需求，通常不需要过多的CPU资源。

JobManager Memory

管理作业的元数据和状态信息

需要大规模状态管理的作业，以及需要长期保存状态信息的情况。

TaskManager CPU

执行作业中的任务。

TaskManager负责执行作业的任务，包括数据处理和计算。

适用于计算密集型的任务，需要大量CPU计算资源。

TaskManager Memory

存储任务数据和状态信息，包括中间结果、缓存数据等。

适用于需要大规模数据存储和中间结果缓存的作业。

每个TaskManager Slot数

每个TaskManager上可并行执行的任务数。

限制了TaskManager上同时执行任务的数量。

适用于控制每个TaskManager的并行度，以避免资源竞争。

指标

指标逻辑

（含义/场景）解释

可能原因

重启次数

Num of Restarts

每个作业重启次数总数

• 代码错误
• 资源不足

业务延时

currentEmitEventTimeLag

数据离开 Source 的时刻（EmitTime）- 数据事件时间（EventTime）

该延时较大时，说明作业可能在拉取数据或者处理数据上存在延时。

• 计算逻辑复杂
• 计算资源不足

传输延时currentFetchEventTimeLag

数据被 Source 读取的时刻（FetchTime）- 数据事件时间（EventTime）

度量数据从数据源到Flink作业的传输延时，识别数据获取和传输过程中的性能瓶颈。

结合 业务延时指标可以分析Source处理能力

• 延迟差值接近：Source从外部系统中拉取数据的能力（网络 I/O、并发数）不足。
• 延迟差值较大：作业的处理能力不足，导致数据在Source中有滞留。

输入记录总数

numRecordsIn

每个 operator（source 端和中间 operator）输入的数据总数

反映了每个operator的输入数据规模，用于监控数据流入作业的速率和分布。

-----

输出记录总数

numRecordsOut

每个operator（source 端和中间 operator）输出的数据总数

监控每个operator的输出数据规模，帮助你了解数据流出作业的速率和分布。

-----

Source 端输入记录总数

numRecordsInOfSource

每个 operator（source 端排除中间 operator）输入的数据总数

用于监控Source端的输入数据规模，有助于了解数据源的性能和数据输入速率。

-----

Sink 端输出记录总数

numRecordsOutOfSink

sink 端（排除中间 operator ）输出的数据总数

用于监控Sink端的输出数据规模，帮助你了解数据写入目标的性能和数据输出速率。

-----

每秒输入记录数numRecordsInPerSecond

每秒输入记录数

用于监控每秒输入数据的速率，帮助你了解数据输入的速度。

-----

每秒输出记录数

numRecordsOutPerSecond

每秒输出记录数

用于监控每秒输出数据的速率，帮助你了解数据输出的速度。

-----

Source 端每秒输入记录

numRecordsInOfSourcePerSecond(IN RPS)

每个 operator（source 端排除中间 operator）输入记录总数/时间（秒）

用于监控Source端每秒的输入速率，有助于了解数据源的性能和数据输入速率。

-----

Sink 端每秒输出记录数

numRecordsOutOfSinkPerSecond (OUT RPS)

每个 operator（sink 端排除中间 operator）输出记录总数/时间（秒）

用于监控Sink端每秒的输出速率，帮助你了解数据写入目标的性能和数据输出速率

-----

源端未读取数据条数pendingRecords

获取时刻位点数据与该时刻上游写入位点数据条数的差值

数据积压，数据处理速度无法赶上数据

-----

源端未处理数据时间sourceIdleTime

当前时间 - 最后一次处理数据的时间

用来评估源端操作符的活动性，帮助你了解源端的工作状态和空闲时间。

-----

告警指标

含义

Restart Count in 1 Minute

1分钟内Job Manager重启次数。

Checkpoint Count in 5 Minutes

5分钟内Checkpoint成功次数。

Emit Delay

业务延时，即数据发生时间与数据离开Source算子的时间差值，单位为秒。

IN RPS

每秒输入记录数

OUT RPS

每秒输出记录数

Source IdIe Time

源端未处理数据的时间

Job Failed

作业失败

告警配置流程

1. 添加联系人、联系人组

1. 配置告警规则

1. 点击确认生效

试用事项分类

具体事项

试用结论

数据同步

1、Flink同步作业启动方式

支持全量、增量及系统检查点启动

2、Flink跨账号数据同步

支持跨账号同步

• 武汉云（内网）：配置VPC，添加白名单
• 双V云（公网）：配置NAT，添加白名单

2、Flink同步数据源支持

Flink多数据源、多账号所涉及的同步链路全部支持

3、Flink同步DDL验证

支持

4、 Flink数据同步准确性

准确

作业运维

1、Flink作业资源分配

基础模式 和 专家模式。

基础模式对绝大多数作业满足需求。

2、Flink作业状态存储

• 快照（手动保存、定时保存）
• Checkpoint（由系统管理）

3、Flink作业延迟处理与失败恢复

不同场景对应不同处理方式。简单描述：
1. 降低作业复杂度

2.减少窗口内数据量

3.增加作业资源配置

4、Flink作业调优模式

• 自动调优
• 定时调优

5、Flink作业告警配置

提供多种指标监控，支持多种通知方式

集群/项目空间管理

1、Flink集群扩缩容/恢复

按量付费不支持变配。

包年包月资源变配不影响线上作业

1、Flink项目代码管理、任务发布

-- 讨论点