Flink CDC同步数据到Apache Iceberg表的异常处理与数据质量保障

一、Flink CDC同步数据到Apache Iceberg表的核心逻辑与应用场景

很多做数据处理的朋友可能会遇到这么个需求：业务系统的数据库里每天产生几十上百万条数据，需要实时同步到一个专门用来做分析的大表中，而且这个大表得支持按时间、地区快速查数据，还能随便改之前的历史数据。这时候很多人会选Flink CDC加Apache Iceberg的组合——Flink CDC专门干“实时抓业务库数据”的活，Apache Iceberg是专门存大表的“数据仓库格式”，能解决传统大表的各种问题。

1.1 核心逻辑拆解

简单说整个流程分三步：第一步，Flink CDC连到业务的数据库（比如MySQL），把数据库里的数据变化（新增、修改、删除）实时抓出来；第二步，Flink把抓出来的数据做简单整理（比如补个时间戳）；第三步，Flink把整理好的数据写到Apache Iceberg的表里面。

举个最常见的应用场景：电商平台的订单数据同步。电商的订单系统用MySQL存订单，每来一个新订单、修改一个订单（比如从待付款改成已付款）、取消一个订单（软删除），都要实时同步到Iceberg表，然后运营团队用这个Iceberg表做实时的订单统计、用户行为分析。这个场景下，要是同步过程出了错，比如漏了某个订单、多写了一次同一个订单，运营看到的统计数据就会错，甚至影响业务决策。

二、同步过程中常见的异常类型与原因

很多人第一次搭这套流程的时候，经常会遇到各种莫名其妙的问题，我整理了最常见的三类异常，基本覆盖了90%的坑。

2.1 数据写入异常

这种异常是最常见的，比如Flink写Iceberg表的时候突然报错中断。常见原因有三个：一是Iceberg表的字段和Flink抓出来的数据字段对不上，比如Iceberg表要求订单金额是数字，Flink传过来的是字符串；二是Iceberg表的分区规则有问题，比如Iceberg要求按订单日期分区，但是Flink传过来的订单日期是空的；三是Iceberg表的存储权限不够，比如Flink用的账号没有往HDFS或者S3写数据的权限。

2.2 数据重复异常

这种异常更隐蔽，很多人直到做数据校验的时候才发现。比如同一个订单号的订单在Iceberg表里出现了两次，甚至多次。常见原因是Flink的任务做了故障恢复，比如Flink任务突然断了，重启之后又把之前已经写过的数据重新写了一遍；还有一种情况是业务数据库的主键设计有问题，比如允许同一个订单号重复插入，Flink抓出来之后就会出现重复数据。

2.3 数据丢失异常

这种异常最麻烦，比如某个时间段的订单数据完全没同步到Iceberg表。常见原因是Flink CDC抓数据的时候，数据库的binlog日志被删了——很多数据库会定期清理binlog，要是Flink任务停了超过binlog的保留时间，重启之后就抓不到之前的binlog了；还有一种情况是Flink的任务并行度设置有问题，导致部分数据没有被正确处理。

三、异常处理的具体方案与代码示例

针对上面的三类异常，我整理了对应的处理方案，每个方案都有完整的代码示例，你可以直接改改参数就能用。

3.1 数据写入异常的处理方案

核心思路是提前做数据校验，要是数据不符合Iceberg表的要求，就把这条数据单独存起来，不要让整个任务中断。具体怎么做呢？我们可以在Flink里加一个“数据校验环节”，校验不通过的话，把数据写到一个专门的错误日志表，然后任务继续跑。

这里我们用的技术栈是Flink 1.17 + Flink CDC 2.4 + Apache Iceberg 1.3，所有代码都是针对这个版本的。

首先是校验规则的代码：

import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.types.Row;

// 订单数据校验函数，输入是Flink CDC抓出来的订单数据（Row类型），输出是合法数据或者错误信息
public class OrderDataValidator extends ProcessFunction<Row, Tuple2<Boolean, Row>> {
    @Override
    public void processElement(Row value, Context ctx, Collector<Tuple2<Boolean, Row>> out) throws Exception {
        // 定义校验规则：订单ID不为空、订单金额是正数、订单日期不为空
        String orderId = (String) value.getField("order_id");
        Double amount = (Double) value.getField("amount");
        String orderDate = (String) value.getField("order_date");

        // 校验逻辑
        if (orderId == null || orderId.isEmpty()) {
            // 校验不通过，输出错误标记和原始数据
            out.collect(Tuple2.of(false, value));
        } else if (amount == null || amount <= 0) {
            out.collect(Tuple2.of(false, value));
        } else if (orderDate == null || orderDate.isEmpty()) {
            out.collect(Tuple2.of(false, value));
        } else {
            // 校验通过，输出合法标记和原始数据
            out.collect(Tuple2.of(true, value));
        }
    }
}

然后是Flink任务的核心代码，把校验环节加进去，合法数据写Iceberg表，非法数据写错误日志表：

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.cdc.connectors.mysql.source.MySqlSource;
import org.apache.flink.cdc.connectors.mysql.table.StartupOptions;
import org.apache.flink.types.Row;
import org.apache.flink.api.java.tuple.Tuple2;

public class FlinkCdcToIcebergWithErrorHandling {
    public static void main(String[] args) throws Exception {
        // 初始化Flink流环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3); // 设置并行度，根据自己的服务器配置调整
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        // 1. 定义MySQL CDC源：抓订单表的数据
        MySqlSource<Row> mysqlSource = MySqlSource.<Row>builder()
                .hostname("127.0.0.1") // MySQL地址
                .port(3306) // MySQL端口
                .databaseList("order_db") // 要抓的数据库名
                .tableList("order_db.order_table") // 要抓的表名
                .username("root") // MySQL账号
                .password("123456") // MySQL密码
                .startupOptions(StartupOptions.initial()) // 启动模式：第一次跑全量同步，之后增量
                .build();

        // 2. 把CDC源转成Flink流，加数据校验环节
        env.fromSource(mysqlSource, WatermarkStrategy.noWatermarks(), "MySQL CDC Source")
                .process(new OrderDataValidator()) // 调用我们写的校验函数
                .filter(tuple -> tuple.f0) // 只保留校验通过的数据
                .map(tuple -> tuple.f1) // 提取合法的Row数据
                // 3. 把合法数据写入Iceberg表
                .addSink(new IcebergSinkFunction("hdfs://localhost:9000/iceberg/warehouse", "order_db", "order_iceberg_table"));

        // 4. 把非法数据写入错误日志表（这里用文件存储，方便后续排查）
        env.fromSource(mysqlSource, WatermarkStrategy.noWatermarks(), "MySQL CDC Source")
                .process(new OrderDataValidator())
                .filter(tuple -> !tuple.f0) // 只保留校验不通过的数据
                .map(tuple -> "错误数据：" + tuple.f1.toString() + "，时间：" + System.currentTimeMillis())
                .writeAsText("/path/to/error/logs/"); // 错误日志的存储路径

        // 启动Flink任务
        env.execute("Flink CDC to Iceberg with Error Handling");
    }
}

3.2 数据重复异常的处理方案

核心思路是用Iceberg表的“行级更新”能力，或者给Flink任务加“去重”逻辑。这里有两种常用的方案，我分别说一下。

第一种方案：利用Iceberg的行级更新。Iceberg表支持按主键更新数据，也就是说，要是Flink写过来的订单数据的主键（比如订单ID）已经存在，Iceberg会自动把旧的数据删掉，只保留最新的那条。这个方案的好处是简单，不用改Flink的代码，只要在创建Iceberg表的时候指定主键就行。

创建Iceberg表的SQL代码（用Flink SQL执行）：

CREATE TABLE iceberg_order_table (
    order_id STRING COMMENT '订单ID，主键',
    user_id STRING COMMENT '用户ID',
    amount DOUBLE COMMENT '订单金额',
    order_date STRING COMMENT '订单日期',
    create_time TIMESTAMP COMMENT '创建时间'
) PARTITIONED BY (order_date) -- 按订单日期分区
WITH (
    'connector' = 'iceberg',
    'catalog-type' = 'hive', -- 这里用Hive作为元数据存储，也可以用Hadoop
    'catalog-name' = 'hive_catalog',
    'warehouse' = 'hdfs://localhost:9000/iceberg/warehouse',
    'table-format' = 'parquet', -- 存储格式用Parquet
    'primary-key' = 'order_id' -- 指定主键为订单ID
);

第二种方案：Flink端去重。要是Iceberg表不能指定主键（比如老版本的Iceberg），我们可以在Flink里加一个去重环节，把同一个订单ID的最新数据留下来。去重的核心逻辑是：按订单ID分组，然后保留最后一条数据。

Flink端去重的代码：

import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.types.Row;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;

// 按订单ID去重的函数，保留最新的订单数据
public class OrderDeduplicationFunction extends KeyedProcessFunction<String, Row, Row> {
    // 状态：存储每个订单ID对应的最新数据
    private ValueState<Row> lastOrderState;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 初始化状态描述符，用来存每个订单的最新数据
        ValueStateDescriptor<Row> stateDescriptor = new ValueStateDescriptor<>("last-order-state", Row.class);
        lastOrderState = getRuntimeContext().getState(stateDescriptor);
    }

    @Override
    public void processElement(Row value, Context ctx, Collector<Row> out) throws Exception {
        // 从Row中提取订单ID和创建时间
        String orderId = (String) value.getField("order_id");
        Long createTime = (Long) value.getField("create_time");

        // 获取当前状态中存储的旧数据
        Row oldOrder = lastOrderState.value();

        // 如果旧数据不存在，或者新数据的创建时间比旧数据晚，就更新状态
        if (oldOrder == null || createTime > (Long) oldOrder.getField("create_time")) {
            lastOrderState.update(value);
            out.collect(value); // 输出最新的数据
        }
        // 否则，不输出旧数据，实现去重
    }
}

然后把这个去重函数加到Flink的流里面，位置在数据校验之后，写入Iceberg之前：

// 2. 把CDC源转成Flink流，加数据校验和去重环节
env.fromSource(mysqlSource, WatermarkStrategy.noWatermarks(), "MySQL CDC Source")
        .process(new OrderDataValidator())
        .filter(tuple -> tuple.f0)
        .map(tuple -> tuple.f1)
        .keyBy(row -> (String) row.getField("order_id")) // 按订单ID分组
        .process(new OrderDeduplicationFunction()) // 调用去重函数
        .addSink(new IcebergSinkFunction(...)); // 写入Iceberg表

3.3 数据丢失异常的处理方案

核心思路是做好备份和校验，比如给Flink CDC抓出来的数据做一个备份，定期校验Iceberg表的数据和备份数据是否一致。

具体怎么做呢？我们可以在Flink里加一个“数据备份环节”，把Flink CDC抓出来的所有数据（不管合法不合法）都写到一个Kafka主题里，然后定期用一个脚本校验Kafka里的数据和Iceberg表的数据是否一致。

首先是Flink端备份数据到Kafka的代码：

import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.api.common.serialization.SimpleStringSchema;

// 初始化Kafka生产者
Properties kafkaProps = new Properties();
kafkaProps.setProperty("bootstrap.servers", "localhost:9092");
FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<>(
        "order_data_backup_topic", // 备份用的Kafka主题
        new SimpleStringSchema(),
        kafkaProps
);

// 把Flink CDC抓出来的数据备份到Kafka
env.fromSource(mysqlSource, WatermarkStrategy.noWatermarks(), "MySQL CDC Source")
        .map(row -> row.toString()) // 把Row转成字符串
        .addSink(kafkaProducer);

然后是定期校验数据的脚本（用Python写的，简单易懂）：

import pykafka
import pyiceberg
from pyiceberg.catalog import load_catalog
from pyiceberg.table import Table

# 1. 连接Kafka，获取备份的订单数据
client = pykafka.KafkaClient(hosts="localhost:9092")
topic = client.topics[b'order_data_backup_topic']
consumer = topic.get_simple_consumer()
kafka_orders = set()
for message in consumer:
    if message is not None:
        order_id = message.value.decode().split(',')[0] # 假设订单数据的第一个字段是订单ID
        kafka_orders.add(order_id)

# 2. 连接Iceberg表，获取同步后的订单数据
catalog = load_catalog("hive", warehouse="hdfs://localhost:9000/iceberg/warehouse")
iceberg_table = catalog.load_table("order_db.order_iceberg_table")
iceberg_orders = set()
for batch in iceberg_table.scan().to_batches():
    for record in batch.to_pylist():
        iceberg_orders.add(record['order_id'])

# 3. 对比两个集合，找出丢失的数据
missing_orders = kafka_orders - iceberg_orders
if len(missing_orders) > 0:
    print(f"发现丢失的订单数据：{missing_orders}")
else:
    print("数据校验通过，没有丢失的数据")

四、数据质量保障的体系化方案

除了上面的异常处理，我们还需要一套体系化的方案来保障数据质量，避免出问题，或者出了问题能及时发现。

4.1 元数据监控

元数据是指表的结构、分区、数据量这些信息，我们可以定期监控Iceberg表的元数据，比如每天检查一次Iceberg表的分区是否完整、数据量是否符合预期。比如我们可以写一个脚本，每天检查Iceberg表的当天分区的数据量，要是数据量比前一天少很多，就发告警。

4.2 数据校验规则固化

把我们之前写的数据校验规则（比如订单ID不为空、订单金额为正数）固化下来，做成一个通用的校验框架，以后不管同步什么表，都可以直接用这个框架，不用每次都重新写校验代码。

4.3 全链路告警

给整个同步链路加告警，比如Flink任务中断了要告警、Iceberg表的数据量异常要告警、校验发现数据丢失要告警。常用的告警工具是Prometheus加Alertmanager，把Flink的监控指标和Iceberg的监控指标都加到Prometheus里，然后设置告警规则。

五、技术优缺点分析与注意事项

5.1 技术优缺点

这套组合的优点很明显：第一，实时性好，能做到秒级同步；第二，Iceberg表的能力强，支持行级更新、分区、快照，适合做数据分析；第三，扩展性好，Flink和Iceberg都能处理PB级的数据。

缺点也有：第一，技术栈比较复杂，涉及到Flink、Flink CDC、Iceberg、MySQL、Kafka等多个组件，需要一定的技术基础；第二，调优难度大，比如Flink的并行度、Iceberg的存储格式、分区规则，都需要根据实际数据量调整；第三，异常处理复杂，一旦出问题，排查起来比较麻烦。

5.2 注意事项

第一，MySQL的binlog要设置合理的保留时间，至少要比Flink任务的最大恢复时间长，比如Flink任务最长可能停3天，那binlog的保留时间就要设成7天；第二，Iceberg表的分区规则要合理，不要太细也不要太粗，太细会导致分区太多，影响查询性能，太粗会导致每个分区的数据量太大，影响写入性能；第三，Flink任务要开启Checkpoint，Checkpoint的间隔要合理，间隔太长会导致任务恢复的时候丢数据，间隔太短会影响性能；第四，要定期清理Iceberg表的旧快照，避免元数据越来越大，影响性能。

六、文章总结

Flink CDC同步数据到Apache Iceberg表是一套非常适合实时数据同步的方案，但是要想用好，必须做好异常处理和数据质量保障。异常处理的核心是提前预判可能出现的问题，针对不同的异常类型做对应的处理；数据质量保障的核心是建立一套体系化的监控、校验、告警机制，及时发现和解决问题。

最后，我建议大家在实际使用的时候，先在测试环境把整个流程跑通，把可能出现的问题都排查一遍，然后再上线到生产环境。上线之后，要定期监控整个链路的运行情况，及时调整参数和规则，保证数据的准确性和稳定性。