S3-17 实时数据接入：Flink CDC 全链路
Flink CDC（Change Data Capture）是智策平台实现实时数据接入的核心管道。通过 Debezium Connector 捕获上游数据库（MySQL / PostgreSQL / Oracle）的 binlog 变更，经过 Flink 流处理引擎进行 Schema 映射、数据清洗和格式转换，最终写入 Iceberg 表并更新 Ontology 实例。本文完整拆解从数据库变更捕获到 Ontology 实例更新的全链路实现，包括 Schema Evolution 处理、Exactly-Once 语义保证和故障恢复机制。
Coomia发布于 2025年7月27日19 分钟阅读
分享本文Twitter / X
S3-17 实时数据接入：Flink CDC 全链路
“系列：S3 数据基座 · 第 17 篇 | 难度：高级 | 阅读时间：20 分钟
#TL;DR
Flink CDC（Change Data Capture）是智策平台实现实时数据接入的核心管道。通过 Debezium Connector 捕获上游数据库（MySQL / PostgreSQL / Oracle）的 binlog 变更，经过 Flink 流处理引擎进行 Schema 映射、数据清洗和格式转换，最终写入 Iceberg 表并更新 Ontology 实例。本文完整拆解从数据库变更捕获到 Ontology 实例更新的全链路实现，包括 Schema Evolution 处理、Exactly-Once 语义保证和故障恢复机制。
#1. 为什么需要 Flink CDC
企业的核心业务数据通常存储在关系型数据库中——ERP 用 Oracle、CRM 用 MySQL、MES 用 PostgreSQL。将这些数据接入 Ontology 平台有两种策略：
批量导入（T+1）：每天凌晨全量或增量同步一次。优点是实现简单，缺点是数据延迟至少一天，无法支持实时决策场景。
实时流式接入（CDC）：实时捕获数据库的每一次变更（INSERT / UPDATE / DELETE），通过流处理管道推送到目标系统。延迟可控制在秒级。
Palantir Foundry 通过 Magritte 组件实现数据集成。智策平台选择 Flink CDC 作为替代方案，原因是：
- 开源成熟：Flink CDC 是 Apache 顶级项目，社区活跃，生产验证充分
- 多数据库支持：原生支持 MySQL、PostgreSQL、Oracle、MongoDB、SQL Server
- Exactly-Once 语义：配合 Flink Checkpoint 和 Iceberg 事务，可实现精确一次处理
- Schema Evolution：支持上游数据库的 DDL 变更自动传播
Code
+------------------------------------------------------------------+
| Flink CDC 全链路数据流 |
| |
| MySQL/PG/Oracle |
| | |
| | binlog / WAL / redo log |
| v |
| Debezium Connector (Flink Source) |
| | |
| | ChangeEvent (JSON / Avro) |
| v |
| Flink Stream Processing |
| |── Schema Mapping (源表字段 → Ontology 属性) |
| |── Data Cleansing (类型转换、空值处理) |
| |── Deduplication (基于主键去重) |
| |── Enrichment (关联维度表补充字段) |
| v |
| Dual Write |
| |── Iceberg Table (通过 Nessie catalog) |
| |── Doris Instance Table (实时查询层) |
| v |
| Ontology Event Bus |
| |── InstanceCreatedEvent |
| |── InstanceUpdatedEvent |
| |── InstanceDeletedEvent |
+------------------------------------------------------------------+
#2. 系统架构
#2.1 三层架构
Flink CDC Pipeline 采用经典的三层架构：Source → Process → Sink。
Code
+------------------------------------------------------------------+
| Flink CDC Pipeline Architecture |
| |
| +-----------------+ +-------------------+ +-----------------+ |
| | Source Layer | | Processing Layer | | Sink Layer | |
| | | | | | | |
| | MySQLSource | | SchemaMapper | | IcebergSink | |
| | PostgresSource | | DataCleanser | | DorisSink | |
| | OracleSource | | Deduplicator | | EventBusSink | |
| | MongoDBSource | | Enricher | | MetricsSink | |
| | | | Router | | | |
| +-----------------+ +-------------------+ +-----------------+ |
| |
| Cross-Cutting Concerns: |
| +------------------------------------------------------------+ |
| | Checkpoint Manager | Schema Registry | Error Handler | Metrics| |
| +------------------------------------------------------------+ |
+------------------------------------------------------------------+
#2.2 Pipeline 配置模型
每条 CDC Pipeline 通过声明式配置定义：
Python
from pydantic import BaseModel, Field
from typing import Optional, List, Dict
from enum import Enum
class SourceType(str, Enum):
MYSQL = "mysql"
POSTGRESQL = "postgresql"
ORACLE = "oracle"
MONGODB = "mongodb"
SQLSERVER = "sqlserver"
class CDCPipelineConfig(BaseModel):
"""CDC Pipeline 配置"""
pipeline_id: str
display_name: str
source: SourceConfig
processing: ProcessingConfig
sink: SinkConfig
checkpoint: CheckpointConfig = CheckpointConfig()
error_handling: ErrorHandlingConfig = ErrorHandlingConfig()
class SourceConfig(BaseModel):
"""数据源配置"""
source_type: SourceType
hostname: str
port: int
database: str
tables: List[str] # 监听的表列表
username: str
password_secret_ref: str # K8s Secret 引用
server_id: Optional[int] = None # MySQL server-id
slot_name: Optional[str] = None # PostgreSQL replication slot
startup_mode: str = "initial" # initial | latest-offset | timestamp
startup_timestamp: Optional[int] = None
class ProcessingConfig(BaseModel):
"""处理层配置"""
schema_mapping: Dict[str, SchemaMapping] # 表名 → 映射规则
deduplication: DeduplicationConfig = DeduplicationConfig()
enrichment: Optional[EnrichmentConfig] = None
class SchemaMapping(BaseModel):
"""Schema 映射：源表 → Ontology ObjectType"""
source_table: str
target_object_type_rid: str
field_mappings: Dict[str, FieldMapping] # 源字段 → 目标属性
primary_key_field: str
rid_template: str = "ri.{object_type}.main.{pk}"
class FieldMapping(BaseModel):
"""字段映射"""
source_column: str
target_property: str
type_conversion: Optional[str] = None # 类型转换函数
default_value: Optional[str] = None
nullable: bool = True
class SinkConfig(BaseModel):
"""输出层配置"""
iceberg_enabled: bool = True
iceberg_catalog: str = "nessie"
iceberg_warehouse: str = "s3://coomia-dip-warehouse/"
doris_enabled: bool = True
doris_fe_endpoints: List[str] = []
event_bus_enabled: bool = True
class CheckpointConfig(BaseModel):
"""Checkpoint 配置"""
interval_ms: int = 60000 # 1 分钟
min_pause_ms: int = 500
timeout_ms: int = 600000 # 10 分钟
max_concurrent: int = 1
state_backend: str = "rocksdb"
state_dir: str = "s3://coomia-dip-checkpoints/"
class ErrorHandlingConfig(BaseModel):
"""错误处理配置"""
max_retries: int = 3
retry_interval_ms: int = 5000
dead_letter_enabled: bool = True
dead_letter_topic: str = "cdc-dead-letter"
skip_corrupted: bool = False
#3. Source Layer：Debezium 变更捕获
#3.1 MySQL CDC Source
MySQL CDC 通过读取 binlog 实现变更捕获。Flink CDC Connector 内嵌了 Debezium 引擎，无需独立部署 Debezium 服务。
Python
class MySQLCDCSource:
"""MySQL CDC Source 构建器"""
def build(self, config: SourceConfig) -> FlinkSource:
"""构建 MySQL CDC Source"""
return MySqlSource.builder() \
.hostname(config.hostname) \
.port(config.port) \
.database_list(config.database) \
.table_list(*config.tables) \
.username(config.username) \
.password(self._resolve_secret(config.password_secret_ref)) \
.server_id(config.server_id or self._generate_server_id()) \
.deserializer(JsonDebeziumDeserializationSchema()) \
.startup_options(self._build_startup_options(config)) \
.build()
def _build_startup_options(self, config: SourceConfig):
if config.startup_mode == "initial":
return StartupOptions.initial()
elif config.startup_mode == "latest-offset":
return StartupOptions.latest_offset()
elif config.startup_mode == "timestamp":
return StartupOptions.timestamp(config.startup_timestamp)
else:
raise ValueError(f"Unknown startup mode: {config.startup_mode}")
#3.2 变更事件结构
Debezium 产生的变更事件（ChangeEvent）包含完整的变更信息：
JSON
{
"before": {
"id": 1001,
"name": "Pump-A",
"status": "running",
"temperature": 72.5
},
"after": {
"id": 1001,
"name": "Pump-A",
"status": "maintenance",
"temperature": 85.3
},
"source": {
"connector": "mysql",
"db": "factory_db",
"table": "equipment",
"ts_ms": 1711267200000,
"server_id": 1,
"file": "mysql-bin.000003",
"pos": 12345
},
"op": "u",
"ts_ms": 1711267200123
}
before：变更前的完整行数据（UPDATE 和 DELETE 时存在）after：变更后的完整行数据（INSERT 和 UPDATE 时存在）op：操作类型——c(create/insert)、u(update)、d(delete)、r(read/snapshot)source：来源信息，包含数据库、表、binlog 位置等
#3.3 全量快照 + 增量同步
Flink CDC 的 initial 启动模式会先执行全量快照，再切换到增量同步：
Code
+------------------------------------------------------------------+
| 全量快照 + 增量同步流程 |
| |
| Phase 1: Snapshot (全量快照) |
| ─────────────────────────── |
| 1. 获取全局读锁（FLUSH TABLES WITH READ LOCK） |
| 2. 记录当前 binlog 位置 |
| 3. 读取表结构（SHOW CREATE TABLE） |
| 4. 释放读锁 |
| 5. 分片读取全量数据（SELECT * FROM table WHERE pk BETWEEN ? AND ?）|
| 6. 每个分片产生 op=r 的事件 |
| |
| Phase 2: Incremental (增量同步) |
| ─────────────────────────── |
| 1. 从快照记录的 binlog 位置开始消费 |
| 2. 实时接收 INSERT/UPDATE/DELETE 事件 |
| 3. 持续运行，直到 Pipeline 停止 |
+------------------------------------------------------------------+
#4. Processing Layer：数据处理
#4.1 Schema Mapper
Schema Mapper 将源表的字段映射到 Ontology 的属性结构：
Python
class SchemaMapper:
"""Schema 映射器：源表字段 → Ontology 属性"""
def __init__(self, mappings: Dict[str, SchemaMapping]):
self._mappings = mappings
def map_event(self, event: ChangeEvent) -> Optional[OntologyChangeEvent]:
"""将 Debezium ChangeEvent 转换为 Ontology ChangeEvent"""
table_name = f"{event.source.db}.{event.source.table}"
mapping = self._mappings.get(table_name)
if mapping is None:
return None # 未配置映射的表，跳过
if event.op in ("c", "r"):
return self._map_create(event, mapping)
elif event.op == "u":
return self._map_update(event, mapping)
elif event.op == "d":
return self._map_delete(event, mapping)
return None
def _map_create(self, event: ChangeEvent, mapping: SchemaMapping) -> OntologyChangeEvent:
properties = {}
for source_col, field_map in mapping.field_mappings.items():
value = event.after.get(source_col)
if value is None and field_map.default_value is not None:
value = field_map.default_value
if field_map.type_conversion:
value = self._convert_type(value, field_map.type_conversion)
properties[field_map.target_property] = value
pk_value = event.after[mapping.primary_key_field]
instance_rid = mapping.rid_template.format(
object_type=mapping.target_object_type_rid.split(".")[-1],
pk=pk_value
)
return OntologyChangeEvent(
operation="create",
object_type_rid=mapping.target_object_type_rid,
instance_rid=instance_rid,
properties=properties,
source_timestamp=event.source.ts_ms,
source_table=f"{event.source.db}.{event.source.table}"
)
def _map_update(self, event: ChangeEvent, mapping: SchemaMapping) -> OntologyChangeEvent:
# 只映射变更的字段
changed_properties = {}
for source_col, field_map in mapping.field_mappings.items():
old_value = event.before.get(source_col) if event.before else None
new_value = event.after.get(source_col) if event.after else None
if old_value != new_value:
if field_map.type_conversion:
new_value = self._convert_type(new_value, field_map.type_conversion)
changed_properties[field_map.target_property] = new_value
pk_value = event.after[mapping.primary_key_field]
instance_rid = mapping.rid_template.format(
object_type=mapping.target_object_type_rid.split(".")[-1],
pk=pk_value
)
return OntologyChangeEvent(
operation="update",
object_type_rid=mapping.target_object_type_rid,
instance_rid=instance_rid,
properties=changed_properties,
source_timestamp=event.source.ts_ms,
source_table=f"{event.source.db}.{event.source.table}"
)
def _convert_type(self, value, conversion: str):
"""类型转换"""
converters = {
"to_string": str,
"to_int": int,
"to_float": float,
"to_boolean": lambda v: v in (1, True, "true", "yes"),
"epoch_ms_to_datetime": lambda v: datetime.fromtimestamp(v / 1000),
"cents_to_yuan": lambda v: v / 100.0 if v else None,
}
converter = converters.get(conversion)
if converter and value is not None:
return converter(value)
return value
#4.2 数据去重（Deduplication）
在 CDC 场景中，由于 Checkpoint 恢复可能导致事件重放，去重是必要的：
Python
class Deduplicator:
"""基于主键和事件时间的去重器"""
def __init__(self, state_backend, dedup_window_ms: int = 60000):
self._state = state_backend
self._window_ms = dedup_window_ms
async def is_duplicate(self, event: OntologyChangeEvent) -> bool:
"""检查事件是否重复"""
key = f"{event.object_type_rid}:{event.instance_rid}"
last_ts = await self._state.get(key)
if last_ts is not None and event.source_timestamp <= last_ts:
return True # 重复事件
await self._state.put(key, event.source_timestamp)
return False
async def cleanup_expired(self):
"""清理过期的去重状态"""
cutoff = int(time.time() * 1000) - self._window_ms
await self._state.remove_before(cutoff)
#4.3 数据丰富（Enrichment）
Enrichment 是在流处理过程中关联维度表，补充额外字段：
Python
class Enricher:
"""数据丰富器：关联维度表补充字段"""
def __init__(self, dimension_sources: Dict[str, DimensionSource]):
self._sources = dimension_sources
async def enrich(
self,
event: OntologyChangeEvent,
enrichment_config: EnrichmentConfig
) -> OntologyChangeEvent:
for rule in enrichment_config.rules:
source = self._sources.get(rule.dimension_source)
if source is None:
continue
lookup_key = event.properties.get(rule.lookup_field)
if lookup_key is None:
continue
dimension_data = await source.lookup(lookup_key)
if dimension_data:
for target_field, source_field in rule.field_mappings.items():
event.properties[target_field] = dimension_data.get(source_field)
return event
#5. Sink Layer：双写 Iceberg + Doris
#5.1 Iceberg Sink
Iceberg Sink 通过 Nessie Catalog 将变更数据写入 Iceberg 表：
Python
class IcebergSinkWriter:
"""Iceberg Sink：写入 Iceberg 表"""
def __init__(self, catalog_config: dict):
self._catalog = NessieCatalog(**catalog_config)
async def write(self, event: OntologyChangeEvent):
table_name = self._resolve_table_name(event.object_type_rid)
table = self._catalog.load_table(table_name)
if event.operation == "create":
record = self._to_iceberg_record(event)
table.append(record)
elif event.operation == "update":
# Iceberg 使用 Merge-on-Read 策略
table.merge(
key=event.instance_rid,
updates=event.properties
)
elif event.operation == "delete":
table.delete(
filter=f"rid = '{event.instance_rid}'"
)
async def commit(self, checkpoint_id: int):
"""在 Checkpoint 时提交 Iceberg 事务"""
self._catalog.commit(
message=f"CDC checkpoint {checkpoint_id}",
branch="main"
)
#5.2 Doris Sink
Doris Sink 通过 Stream Load 接口实时写入，支持 UPSERT 语义：
Python
class DorisSinkWriter:
"""Doris Sink：Stream Load 写入"""
def __init__(self, fe_endpoints: List[str]):
self._endpoints = fe_endpoints
self._buffer: List[dict] = []
self._buffer_size = 1000
self._flush_interval_ms = 5000
async def write(self, event: OntologyChangeEvent):
row = self._to_doris_row(event)
self._buffer.append(row)
if len(self._buffer) >= self._buffer_size:
await self._flush()
async def _flush(self):
if not self._buffer:
return
table_name = self._resolve_table_name(self._buffer[0])
csv_data = self._to_csv(self._buffer)
endpoint = self._select_endpoint()
response = await self._stream_load(
endpoint=endpoint,
table=table_name,
data=csv_data,
format="csv",
merge_type="MERGE", # UPSERT 语义
delete_condition="__deleted = 1"
)
if response.status != "Success":
raise DorisSinkError(f"Stream Load failed: {response.message}")
self._buffer.clear()
async def _stream_load(self, endpoint, table, data, **kwargs):
"""调用 Doris Stream Load API"""
url = f"http://{endpoint}/api/{table}/_stream_load"
headers = {
"Content-Type": "text/csv",
"format": kwargs.get("format", "csv"),
"merge_type": kwargs.get("merge_type", "APPEND"),
}
async with aiohttp.ClientSession() as session:
async with session.put(url, data=data, headers=headers) as resp:
return await resp.json()
#5.3 双写一致性
Iceberg 和 Doris 的双写需要保证一致性。采用的策略是：以 Iceberg 为主（真实数据源），Doris 为副（查询加速层）。
Code
+------------------------------------------------------------------+
| 双写一致性策略 |
| |
| Flink Checkpoint |
| | |
| v |
| Phase 1: Pre-commit |
| |── Iceberg: 准备提交（写入 data files） |
| |── Doris: 缓冲区准备就绪 |
| v |
| Phase 2: Commit |
| |── Iceberg: 提交事务（原子操作） |
| |── Doris: Stream Load 提交 |
| v |
| Phase 3: Post-commit |
| |── 发布事件到 Event Bus |
| |── 更新 Pipeline 状态 |
| |
| 故障恢复： |
| 如果 Doris 写入失败但 Iceberg 成功： |
| → 触发 Doris 补偿任务从 Iceberg 重新加载数据 |
+------------------------------------------------------------------+
#6. Schema Evolution 处理
上游数据库的表结构变更（DDL）是 CDC Pipeline 的重要挑战。Flink CDC 支持自动捕获 DDL 事件并传播。
Python
class SchemaEvolutionHandler:
"""Schema Evolution 处理器"""
def __init__(self, schema_registry, iceberg_catalog):
self._schema_registry = schema_registry
self._iceberg = iceberg_catalog
async def handle_ddl(self, ddl_event: DDLEvent):
"""处理 DDL 变更事件"""
if ddl_event.type == "ALTER_TABLE_ADD_COLUMN":
await self._handle_add_column(ddl_event)
elif ddl_event.type == "ALTER_TABLE_MODIFY_COLUMN":
await self._handle_modify_column(ddl_event)
elif ddl_event.type == "ALTER_TABLE_DROP_COLUMN":
await self._handle_drop_column(ddl_event)
async def _handle_add_column(self, ddl_event: DDLEvent):
"""处理新增列"""
mapping = self._get_mapping(ddl_event.table)
if mapping is None:
return
# 1. 更新 Ontology Schema（注册新属性）
new_property = PropertyDefinition(
api_name=ddl_event.column_name,
display_name=ddl_event.column_name,
property_type=self._map_sql_type(ddl_event.column_type),
nullable=ddl_event.nullable
)
await self._schema_registry.add_property(
mapping.target_object_type_rid,
new_property
)
# 2. 更新 Iceberg 表结构
table = self._iceberg.load_table(
self._resolve_table_name(mapping.target_object_type_rid)
)
table.update_schema() \
.add_column(ddl_event.column_name, self._to_iceberg_type(ddl_event.column_type)) \
.commit()
# 3. 更新字段映射
mapping.field_mappings[ddl_event.column_name] = FieldMapping(
source_column=ddl_event.column_name,
target_property=ddl_event.column_name
)
async def _handle_modify_column(self, ddl_event: DDLEvent):
"""处理列类型变更"""
mapping = self._get_mapping(ddl_event.table)
if mapping is None:
return
# Iceberg 支持安全的类型提升（int → long, float → double）
table = self._iceberg.load_table(
self._resolve_table_name(mapping.target_object_type_rid)
)
old_type = table.schema().find_field(ddl_event.column_name).field_type
new_type = self._to_iceberg_type(ddl_event.column_type)
if self._is_safe_promotion(old_type, new_type):
table.update_schema() \
.update_column(ddl_event.column_name, new_type) \
.commit()
else:
# 不安全的类型变更：记录警告，需要人工介入
await self._alert_unsafe_schema_change(ddl_event)
#7. Exactly-Once 语义保证
#7.1 端到端 Exactly-Once
Flink CDC 的 Exactly-Once 语义通过三个组件协作实现：
Code
+------------------------------------------------------------------+
| Exactly-Once 语义实现 |
| |
| 组件 机制 |
| ────────── ────────────────────── |
| Source (Debezium) binlog offset 持久化到 Flink State |
| Processing (Flink) Checkpoint Barrier 对齐 |
| Sink (Iceberg) 两阶段提交 (2PC) |
| Sink (Doris) 幂等 UPSERT (基于主键去重) |
| |
| Checkpoint 流程： |
| 1. JobManager 注入 Checkpoint Barrier |
| 2. Source 记录当前 binlog offset |
| 3. Barrier 流过所有算子，每个算子持久化本地状态 |
| 4. Sink 执行 Pre-commit（写入但不提交） |
| 5. 所有算子完成后，JobManager 通知 Commit |
| 6. Sink 执行 Commit（原子提交） |
+------------------------------------------------------------------+
#7.2 故障恢复
当 Pipeline 发生故障时，从最近的 Checkpoint 恢复：
Python
class CDCPipelineRecovery:
"""CDC Pipeline 故障恢复"""
async def recover(self, pipeline_id: str):
"""从最近的 Checkpoint 恢复 Pipeline"""
# 1. 获取最近的 Checkpoint
checkpoint = await self._checkpoint_store.get_latest(pipeline_id)
if checkpoint is None:
raise RecoveryError(f"No checkpoint found for pipeline {pipeline_id}")
# 2. 恢复 Source 的 binlog 位置
binlog_offset = checkpoint.source_state["binlog_offset"]
# 3. 恢复 Processing 的状态（去重状态等）
processing_state = checkpoint.processing_state
# 4. 恢复 Sink 的事务状态
# Iceberg: 回滚未提交的事务
await self._iceberg_sink.rollback_pending()
# Doris: 幂等重放，无需特殊处理
# 5. 从 Checkpoint 位置重启 Pipeline
await self._restart_pipeline(
pipeline_id,
binlog_offset=binlog_offset,
processing_state=processing_state
)
async def _restart_pipeline(self, pipeline_id, binlog_offset, processing_state):
config = await self._config_store.get(pipeline_id)
config.source.startup_mode = "specific-offset"
config.source.startup_offset = binlog_offset
pipeline = CDCPipelineBuilder(config).build()
pipeline.restore_state(processing_state)
await pipeline.start()
#8. 监控与运维
#8.1 Pipeline 监控指标
Python
class CDCPipelineMetrics:
"""CDC Pipeline 监控指标"""
def __init__(self, metrics_registry):
self._registry = metrics_registry
# Source 指标
self.source_events_total = self._registry.counter(
"cdc_source_events_total",
labels=["pipeline_id", "table", "operation"]
)
self.source_lag_ms = self._registry.gauge(
"cdc_source_lag_ms",
labels=["pipeline_id"]
)
# Processing 指标
self.processing_latency_ms = self._registry.histogram(
"cdc_processing_latency_ms",
labels=["pipeline_id", "stage"]
)
self.dedup_hit_total = self._registry.counter(
"cdc_dedup_hit_total",
labels=["pipeline_id"]
)
# Sink 指标
self.sink_write_total = self._registry.counter(
"cdc_sink_write_total",
labels=["pipeline_id", "sink_type", "status"]
)
self.sink_latency_ms = self._registry.histogram(
"cdc_sink_latency_ms",
labels=["pipeline_id", "sink_type"]
)
# Checkpoint 指标
self.checkpoint_duration_ms = self._registry.histogram(
"cdc_checkpoint_duration_ms",
labels=["pipeline_id"]
)
self.checkpoint_size_bytes = self._registry.gauge(
"cdc_checkpoint_size_bytes",
labels=["pipeline_id"]
)
#8.2 延迟告警
CDC Pipeline 的延迟是最关键的监控指标。延迟定义为：当前时间 - 事件在源数据库中的发生时间。
Python
class CDCLagMonitor:
"""CDC 延迟监控"""
def __init__(self, alert_threshold_ms: int = 30000):
self._threshold_ms = alert_threshold_ms
async def check_lag(self, pipeline_id: str, event: OntologyChangeEvent):
current_ms = int(time.time() * 1000)
lag_ms = current_ms - event.source_timestamp
metrics.source_lag_ms.set(lag_ms, pipeline_id=pipeline_id)
if lag_ms > self._threshold_ms:
await self._send_alert(
pipeline_id=pipeline_id,
lag_ms=lag_ms,
threshold_ms=self._threshold_ms,
message=f"CDC pipeline {pipeline_id} lag {lag_ms}ms exceeds threshold {self._threshold_ms}ms"
)
#9. Pipeline 生命周期管理
#9.1 Pipeline 管理 API
Python
class CDCPipelineManager:
"""CDC Pipeline 生命周期管理"""
async def create(self, config: CDCPipelineConfig) -> str:
"""创建 Pipeline"""
# 验证配置
await self._validate_config(config)
# 验证数据源连通性
await self._test_connection(config.source)
# 注册 Pipeline
pipeline_id = await self._store.save(config)
return pipeline_id
async def start(self, pipeline_id: str):
"""启动 Pipeline"""
config = await self._store.get(pipeline_id)
pipeline = CDCPipelineBuilder(config).build()
await pipeline.start()
await self._store.update_status(pipeline_id, "RUNNING")
async def stop(self, pipeline_id: str):
"""停止 Pipeline（保存 Checkpoint）"""
pipeline = self._running.get(pipeline_id)
if pipeline:
await pipeline.stop_with_savepoint()
await self._store.update_status(pipeline_id, "STOPPED")
async def restart(self, pipeline_id: str):
"""重启 Pipeline（从上次 Checkpoint 恢复）"""
await self.stop(pipeline_id)
await self.start(pipeline_id)
async def get_status(self, pipeline_id: str) -> PipelineStatus:
"""获取 Pipeline 状态"""
config = await self._store.get(pipeline_id)
metrics = await self._metrics.get_pipeline_metrics(pipeline_id)
return PipelineStatus(
pipeline_id=pipeline_id,
status=config.status,
source_lag_ms=metrics.source_lag_ms,
events_per_second=metrics.events_per_second,
last_checkpoint_at=metrics.last_checkpoint_at,
error_count=metrics.error_count
)
#10. 生产部署配置
#10.1 Kubernetes 部署
YAML
apiVersion: apps/v1
kind: Deployment
metadata:
name: flink-cdc-jobmanager
namespace: coomia-dip
spec:
replicas: 1
template:
spec:
containers:
- name: flink-jobmanager
image: coomia-dip/flink-cdc:1.18
args: ["jobmanager"]
resources:
requests:
memory: "2Gi"
cpu: "1"
limits:
memory: "4Gi"
cpu: "2"
env:
- name: FLINK_PROPERTIES
value: |
jobmanager.rpc.address: flink-cdc-jobmanager
state.backend: rocksdb
state.checkpoints.dir: s3://coomia-dip-checkpoints/
execution.checkpointing.interval: 60000
execution.checkpointing.min-pause: 500
---
apiVersion: apps/v1
kind: Deployment
metadata:
name: flink-cdc-taskmanager
namespace: coomia-dip
spec:
replicas: 3
template:
spec:
containers:
- name: flink-taskmanager
image: coomia-dip/flink-cdc:1.18
args: ["taskmanager"]
resources:
requests:
memory: "4Gi"
cpu: "2"
limits:
memory: "8Gi"
cpu: "4"
#10.2 资源规划
Code
+------------------------------------------------------------------+
| 资源规划参考 |
| |
| 源数据库规模 TPS TaskManager 数 内存/TM 并行度 |
| ────────────── ───── ────────────── ──────── ──────── |
| 小型（<50 表） <1K 2 4GB 4 |
| 中型（50-200 表） 1K-5K 4 8GB 8 |
| 大型（>200 表） >5K 8+ 16GB 16 |
+------------------------------------------------------------------+
#11. 与 Palantir 数据集成对比
Code
+------------------------------------------------------------------+
| 数据集成能力对比 |
| |
| 能力 Palantir Magritte 智策 Flink CDC |
| ────────────── ────────────────── ────────────── |
| CDC 支持 ✅ (闭源) ✅ (开源) |
| MySQL/PG/Oracle ✅ ✅ |
| MongoDB ✅ ✅ |
| 批量导入 ✅ ✅ |
| Schema Evolution ✅ ✅ |
| Exactly-Once ✅ ✅ |
| 延迟 秒级 秒级 |
| 自定义转换 ✅ (代码) ✅ (配置+代码) |
| 可视化编排 ✅ 🔄 (规划中) |
| 开源可控 ❌ ✅ |
+------------------------------------------------------------------+
#12. 测试策略
Python
class TestCDCPipeline:
@pytest.fixture
async def mysql_container(self):
"""启动 MySQL Testcontainer"""
async with MySQLContainer("mysql:8.0") as mysql:
yield mysql
@pytest.mark.asyncio
async def test_full_cdc_pipeline(self, mysql_container):
"""端到端测试：MySQL 变更 → Ontology 实例"""
# 1. 创建源表并插入数据
await mysql_container.execute("""
CREATE TABLE equipment (
id INT PRIMARY KEY,
name VARCHAR(100),
status VARCHAR(50)
)
""")
await mysql_container.execute(
"INSERT INTO equipment VALUES (1, 'Pump-A', 'running')"
)
# 2. 启动 CDC Pipeline
config = CDCPipelineConfig(
pipeline_id="test-pipeline",
source=SourceConfig(
source_type=SourceType.MYSQL,
hostname=mysql_container.host,
port=mysql_container.port,
database="test",
tables=["test.equipment"],
username="root",
password_secret_ref="test-secret"
),
processing=ProcessingConfig(
schema_mapping={
"test.equipment": SchemaMapping(
source_table="test.equipment",
target_object_type_rid="ri.type.Equipment",
field_mappings={
"id": FieldMapping(source_column="id", target_property="equipment_id"),
"name": FieldMapping(source_column="name", target_property="display_name"),
"status": FieldMapping(source_column="status", target_property="status")
},
primary_key_field="id"
)
}
),
sink=SinkConfig(iceberg_enabled=False, doris_enabled=False, event_bus_enabled=True)
)
pipeline = CDCPipelineBuilder(config).build()
events = []
pipeline.add_listener(lambda e: events.append(e))
await pipeline.start()
# 3. 等待快照完成
await asyncio.sleep(5)
# 4. 验证初始快照事件
assert len(events) == 1
assert events[0].operation == "create"
assert events[0].properties["display_name"] == "Pump-A"
# 5. 执行 UPDATE
await mysql_container.execute(
"UPDATE equipment SET status = 'maintenance' WHERE id = 1"
)
await asyncio.sleep(2)
# 6. 验证增量变更事件
assert len(events) == 2
assert events[1].operation == "update"
assert events[1].properties["status"] == "maintenance"
await pipeline.stop()
#Key Takeaways
- Flink CDC 是智策平台实时数据接入的核心管道，通过 Debezium 捕获数据库 binlog 变更，实现秒级延迟的实时同步
- **三层架构（Source → Process → Sink）**提供了清晰的关注点分离，每层独立可配置、可扩展
- Schema Mapper 将源表字段映射到 Ontology 属性结构，支持类型转换、默认值和丰富化
- 双写 Iceberg + Doris 兼顾了数据湖的版本化存储和 OLAP 的实时查询能力
- Exactly-Once 语义 通过 Flink Checkpoint + Iceberg 2PC + Doris 幂等 UPSERT 三方协作实现
- Schema Evolution 自动处理上游 DDL 变更，安全的类型提升自动传播，不安全的变更触发告警
#Next Article
下一篇 S3-18 Pipeline DSL 设计：Python 链式 API 将深入剖析智策平台的 Pipeline DSL 设计，如何通过 Python 链式 API 声明式定义数据处理管道。
“Tags:
flink-cdcchange-data-capturedebeziumreal-time-ingestionbinlogicebergdorisexactly-onceschema-evolutioncoomia-dip