不支持标准的Debezium JSON吗

[benxiaohai061]

我现在想实现一个实时同步mysql数据到doris的方案(整库同步）。但由于网络隔离原因。 mysql和doris不在同一个网络。我使用了如下方案。
1、通过flink-cdc 实时把mysql的数据生成Debezium JSON格式文件，
2、然后再通过FTP传入内网。
3、再使用flink程序实时读取内网接收的文件入doris。

但现在出现的问题是。
1、doris的表必须存在。 不能自动建表。
2、标准Debezium JSON的格式，doris sink执行会报错。
15:51:34,285 WARN org.apache.doris.flink.sink.writer.serializer.jsondebezium.JsonDebeziumSchemaChange [] - Failed to extract tableChanges. record={"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"student_id"},{"type":"string","optional":false,"field":"first_name"},{"type":"string","optional":false,"field":"last_name"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"Male,Female,Other"},"field":"gender"},{"type":"string","optional":true,"field":"date_of_birth"},{"type":"string","optional":true,"field":"email"},{"type":"string","optional":true,"field":"phone_number"},{"type":"string","optional":true,"field":"address"},{"type":"string","optional":true,"field":"enrollment_date"},{"type":"int16","optional":true,"default":1,"field":"is_active"}],"optional":true,"name":"mysql_binlog_source.test.students.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"student_id"},{"type":"string","optional":false,"field":"first_name"},{"type":"string","optional":false,"field":"last_name"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"Male,Female,Other"},"field":"gender"},{"type":"string","optional":true,"field":"date_of_birth"},{"type":"string","optional":true,"field":"email"},{"type":"string","optional":true,"field":"phone_number"},{"type":"string","optional":true,"field":"address"},{"type":"string","optional":true,"field":"enrollment_date"},{"type":"int16","optional":true,"default":1,"field":"is_active"}],"optional":true,"name":"mysql_binlog_source.test.students.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"string","optional":true,"field":"table"},{"type":"int64","optional":false,"field":"server_id"},{"type":"string","optional":true,"field":"gtid"},{"type":"string","optional":false,"field":"file"},{"type":"int64","optional":false,"field":"pos"},{"type":"int32","optional":false,"field":"row"},{"type":"int64","optional":true,"field":"thread"},{"type":"string","optional":true,"field":"query"}],"optional":false,"name":"io.debezium.connector.mysql.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"field":"transaction"}],"optional":false,"name":"mysql_binlog_source.test.students.Envelope"},"payload":{"before":null,"after":{"student_id":5,"first_name":"Charlie","last_name":"Davis","gender":"Other","date_of_birth":"2002-07-30","email":"charlie.d@example.com","phone_number":null,"address":null,"enrollment_date":"2023-09-01","is_active":0},"source":{"version":"1.9.8.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":0,"snapshot":"false","db":"test","sequence":null,"table":"students","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1760686786704,"transaction":null}}
15:51:34,286 WARN org.apache.doris.flink.sink.writer.serializer.jsondebezium.JsonDebeziumSchemaChangeImplV2 [] - Failed to parse eventType. recordRoot={"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"student_id"},{"type":"string","optional":false,"field":"first_name"},{"type":"string","optional":false,"field":"last_name"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"Male,Female,Other"},"field":"gender"},{"type":"string","optional":true,"field":"date_of_birth"},{"type":"string","optional":true,"field":"email"},{"type":"string","optional":true,"field":"phone_number"},{"type":"string","optional":true,"field":"address"},{"type":"string","optional":true,"field":"enrollment_date"},{"type":"int16","optional":true,"default":1,"field":"is_active"}],"optional":true,"name":"mysql_binlog_source.test.students.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"student_id"},{"type":"string","optional":false,"field":"first_name"},{"type":"string","optional":false,"field":"last_name"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"Male,Female,Other"},"field":"gender"},{"type":"string","optional":true,"field":"date_of_birth"},{"type":"string","optional":true,"field":"email"},{"type":"string","optional":true,"field":"phone_number"},{"type":"string","optional":true,"field":"address"},{"type":"string","optional":true,"field":"enrollment_date"},{"type":"int16","optional":true,"default":1,"field":"is_active"}],"optional":true,"name":"mysql_binlog_source.test.students.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"string","optional":true,"field":"table"},{"type":"int64","optional":false,"field":"server_id"},{"type":"string","optional":true,"field":"gtid"},{"type":"string","optional":false,"field":"file"},{"type":"int64","optional":false,"field":"pos"},{"type":"int32","optional":false,"field":"row"},{"type":"int64","optional":true,"field":"thread"},{"type":"string","optional":true,"field":"query"}],"optional":false,"name":"io.debezium.connector.mysql.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"field":"transaction"}],"optional":false,"name":"mysql_binlog_source.test.students.Envelope"},"payload":{"before":null,"after":{"student_id":5,"first_name":"Charlie","last_name":"Davis","gender":"Other","date_of_birth":"2002-07-30","email":"charlie.d@example.com","phone_number":null,"address":null,"enrollment_date":"2023-09-01","is_active":0},"source":{"version":"1.9.8.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":0,"snapshot":"false","db":"test","sequence":null,"table":"students","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1760686786704,"transaction":null}}
15:51:34,286 W

相关代码是：
1、实时生成文件
Map<String, Object> customConverterConfigs = new HashMap<>();
customConverterConfigs.put(JsonConverterConfig.DECIMAL_FORMAT_CONFIG, "numeric");
JsonDebeziumDeserializationSchema schema =
new JsonDebeziumDeserializationSchema(true, customConverterConfigs);

    MySqlSource<String> mySqlSource =
            MySqlSource.<String>builder()
                    .hostname("127.0.0.1")
                    .port(23306)
                    .databaseList("test") // set captured database
                    .tableList("test\\..*")// set captured table
                    .username("root")
                    .password("123456")
                    .debeziumProperties(DateToStringConverter.DEFAULT_PROPS)
                    .deserializer(schema)
                    .serverTimeZone("Asia/Shanghai")
                    .includeSchemaChanges(true) // converts SourceRecord to JSON String
                    .build();

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    // 设置 3s 的 checkpoint 间隔
    env.enableCheckpointing(3000);

    DataStream<String> cdcStream = env
            .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
            .setParallelism(2);

    // 3. 文件系统 Sink
    FileSink<String> fileSink = FileSink
            .forRowFormat(
                    new Path("D:/data/mysql_cdc/"),          // 输出路径
                    new SimpleStringEncoder<String>("UTF-8"))
            .withRollingPolicy(
                    DefaultRollingPolicy.builder()
                            .withRolloverInterval(Duration.ofMinutes(1))   // 每分钟滚动
                            .withInactivityInterval(Duration.ofSeconds(30)) // 空闲超过30秒滚动
                            .withMaxPartSize(1024 * 1024 * 128)            // 128MB
                            .build()
            )
            .build();

    // 4. 写入文件
    cdcStream.sinkTo(fileSink).setParallelism(1);

    env.execute("MySQL CDC to File");

2、写入doris
public static void main(String[] args) throws Exception {

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    // 开启 checkpoint
    env.enableCheckpointing(10000);

    // ----------------------
    // 1. FileSource 读取 MySQL CDC JSON 文件
    // ----------------------
    FileSource<String> fileSource = FileSource
            .forRecordStreamFormat(new TextLineInputFormat(), new Path("D:/data/mysql_cdc/"))
            .monitorContinuously(Duration.ofSeconds(5)) // 每5秒扫描一次新文件
            .build();

    DataStream<String> fileStream = env.fromSource(
            fileSource,
            WatermarkStrategy.noWatermarks(),
            "File Source"
    );

    // ----------------------
    // 2. Doris Sink 配置
    // ----------------------
    Properties props = new Properties();
    props.setProperty("format", "json");
    props.setProperty("read_json_by_line", "true");

    DorisOptions dorisOptions = DorisOptions.builder()
            .setFenodes("127.0.0.1:8030")
            .setTableIdentifier("")
            .setUsername("root")
            .setPassword("123456")
            .build();

    DorisExecutionOptions.Builder executionBuilder = DorisExecutionOptions.builder();
    executionBuilder
            .setLabelPrefix("label-doris" + UUID.randomUUID())
            .setStreamLoadProp(props)
            .setDeletable(true);

    DorisSink.Builder<String> builder = DorisSink.builder();
    builder.setDorisReadOptions(DorisReadOptions.builder().build())
            .setDorisExecutionOptions(executionBuilder.build())
            .setDorisOptions(dorisOptions)
            .setSerializer(
                    JsonDebeziumSchemaSerializer.builder()
                            .setDorisOptions(dorisOptions)
                            .setNewSchemaChange(true)
                            .build());

    // ----------------------
    // 3. 写入 Doris
    // ----------------------
    fileStream/*.map(new DebeziumToDorisJsonMapper())*/.sinkTo(builder.build());

    env.execute("File to Doris");
}

[kaori-seasons]

您好，我认真看完了您的问题,我梳理了一下我的方案

1.1 错误的核心原因

您遇到的问题本质在于：通过文件中转破坏了CDC事件流的完整性。

从错误日志可以看到，这是一个数据变更事件（op="r" 表示snapshot read操作），但代码尝试从中提取 tableChanges 字段。问题在于

tableChanges 字段只存在于DDL schema change事件中，普通的数据变更事件（INSERT/UPDATE/DELETE/READ）不包含此字段。

1.2 事件区分机制

JsonDebeziumSchemaSerializer 通过以下逻辑区分事件类型：

• 有 op 字段 → 数据变更事件（INSERT/UPDATE/DELETE/READ）
• 无 op 字段 → Schema change事件（DDL）

您的方案将两种事件混在同一个文件中，导致处理逻辑混乱。

1.3 架构缺陷

graph LR
    A[MySQL CDC Source] -->|包含DDL+DML| B[写入文件]
    B --> C[FTP传输]
    C --> D[内网读取文件]
    D -->|无法区分事件类型| E[Doris Sink报错]
    
    style E fill:#f88,stroke:#f00

Loading

---

二、生产可用解决方案（推荐优先级排序）

方案A：通过跳板机实现直连

这是最佳方案，使用Doris-Flink-Connector内置的 DatabaseSync 工具进行整库同步。

A.1 架构设计

graph TB
    subgraph 外网区域
        MySQL[(MySQL)]
    end
    
    subgraph 跳板机/代理
        Proxy[SSH Tunnel/Socks代理]
    end
    
    subgraph 内网区域
        Flink[Flink Job<br/>DatabaseSync]
        Doris[(Doris)]
    end
    
    MySQL -.->|SSH/VPN| Proxy
    Proxy -->|转发| Flink
    Flink -->|直写| Doris

Loading

A.2 实现代码

使用 MysqlDatabaseSync 工具类：
完整代码示例（通过SSH隧道）：

public class MySQLToDorisSync {
    public static void main(String[] args) throws Exception {
        // 1. 先建立SSH隧道映射MySQL端口
        // ssh -L 3307:mysql-host:3306 user@jump-server
        
        // 2. 配置同步参数
        Map<String, String> config = new HashMap<>();
        
        // MySQL配置（通过隧道连接）
        config.put("mysql-conf.hostname", "localhost");  // 通过隧道
        config.put("mysql-conf.port", "3307");           // 本地映射端口
        config.put("mysql-conf.username", "root");
        config.put("mysql-conf.password", "123456");
        config.put("mysql-conf.database-name", "test");
        config.put("mysql-conf.scan-startup-mode", "initial");
        
        // Doris配置
        config.put("sink.fenodes", "127.0.0.1:8030");
        config.put("sink.username", "root");
        config.put("sink.password", "123456");
        config.put("sink.jdbc-url", "jdbc:mysql://127.0.0.1:9030");
        
        // 同步配置
        config.put("sink.sink.batch.size", "10000");
        config.put("sink.sink.max-retries", "3");
        config.put("sink.sink.batch.interval", "10s");
        
        // 表过滤（可选）
        config.put("including-tables", "test\\..*");
        
        // Schema change模式（推荐SQL_PARSER）
        config.put("schema-change-mode", "sql_parser");
        
        // 3. 创建并执行同步任务
        Configuration configuration = Configuration.fromMap(config);
        MysqlDatabaseSync sync = new MysqlDatabaseSync();
        sync.setConfig(configuration);
        sync.create();
        sync.build();
    }
}

A.3 网络连接方案

选项1：SSH隧道

# 在内网Flink机器上执行
ssh -N -L 3307:mysql-real-host:3306 user@jump-server

选项2：SOCKS代理

// 在Flink任务启动参数中设置
-DsocksProxyHost=proxy-host
-DsocksProxyPort=1080

选项3：VPN专线（最稳定）

• 企业级VPN打通网络
• 直接使用真实MySQL地址

A.4 核心优势

• 自动建表：根据MySQL schema自动在Doris创建表
• Schema演化：自动同步DDL变更（ADD/DROP/MODIFY COLUMN）
• 事务一致性：保证数据一致性
• 断点续传：支持checkpoint恢复
• 生产稳定：官方推荐方案

---

方案B：改进文件传输方案

如果无法建立直连，可以优化文件传输架构。

B.1 核心思路：分离Schema Change和Data Change

graph LR
    A[MySQL CDC] -->|判断事件类型| B{有op字段?}
    B -->|是| C[数据变更文件]
    B -->|否| D[DDL变更文件]
    C --> E[FTP传输]
    D --> E
    E --> F[内网处理]
    F -->|先处理DDL| G[创建/修改表]
    F -->|后处理数据| H[写入数据]

Loading

B.2 实现代码

步骤1：生成文件时分离事件

public class MySQLCDCToFileWithSeparation {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(3000);

        // 配置CDC Source
        Map<String, Object> customConverterConfigs = new HashMap<>();
        customConverterConfigs.put(JsonConverterConfig.DECIMAL_FORMAT_CONFIG, "numeric");
        JsonDebeziumDeserializationSchema schema =
                new JsonDebeziumDeserializationSchema(true, customConverterConfigs);

        MySqlSource<String> mySqlSource =
                MySqlSource.<String>builder()
                        .hostname("127.0.0.1")
                        .port(23306)
                        .databaseList("test")
                        .tableList("test\\..*")
                        .username("root")
                        .password("123456")
                        .debeziumProperties(DateToStringConverter.DEFAULT_PROPS)
                        .deserializer(schema)
                        .serverTimeZone("Asia/Shanghai")
                        .includeSchemaChanges(true)
                        .build();

        DataStream<String> cdcStream = env
                .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
                .setParallelism(2);

        // 分流：根据是否有op字段判断
        OutputTag<String> ddlTag = new OutputTag<String>("ddl-events"){};
        
        SingleOutputStreamOperator<String> dataStream = cdcStream.process(
            new ProcessFunction<String, String>() {
                @Override
                public void processElement(String value, Context ctx, Collector<String> out) {
                    try {
                        JsonNode node = new ObjectMapper().readTree(value);
                        if (node.has("op") && node.get("op") != null) {
                            // 数据变更事件
                            out.collect(value);
                        } else {
                            // DDL事件
                            ctx.output(ddlTag, value);
                        }
                    } catch (Exception e) {
                        // 容错处理
                        out.collect(value);
                    }
                }
            }
        );

        // DDL事件单独输出
        DataStream<String> ddlStream = dataStream.getSideOutput(ddlTag);

        // 1. 数据变更写入data目录
        FileSink<String> dataSink = FileSink
                .forRowFormat(
                        new Path("D:/data/mysql_cdc/data/"),
                        new SimpleStringEncoder<String>("UTF-8"))
                .withRollingPolicy(
                        DefaultRollingPolicy.builder()
                                .withRolloverInterval(Duration.ofMinutes(1))
                                .withInactivityInterval(Duration.ofSeconds(30))
                                .withMaxPartSize(1024 * 1024 * 128)
                                .build())
                .build();

        // 2. DDL事件写入ddl目录
        FileSink<String> ddlSink = FileSink
                .forRowFormat(
                        new Path("D:/data/mysql_cdc/ddl/"),
                        new SimpleStringEncoder<String>("UTF-8"))
                .withRollingPolicy(
                        DefaultRollingPolicy.builder()
                                .withRolloverInterval(Duration.ofMinutes(5))
                                .withInactivityInterval(Duration.ofSeconds(60))
                                .build())
                .build();

        dataStream.sinkTo(dataSink).setParallelism(1);
        ddlStream.sinkTo(ddlSink).setParallelism(1);

        env.execute("MySQL CDC to Separated Files");
    }
}

步骤2：内网分别处理

public class SeparatedFilesToDoris {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(10000);

        // 配置Doris连接
        DorisOptions dorisOptions = DorisOptions.builder()
                .setFenodes("127.0.0.1:8030")
                .setTableIdentifier("")
                .setUsername("root")
                .setPassword("123456")
                .build();

        // ========== 任务1：处理DDL事件（优先级高） ==========
        FileSource<String> ddlSource = FileSource
                .forRecordStreamFormat(new TextLineInputFormat(), 
                                      new Path("D:/data/mysql_cdc/ddl/"))
                .monitorContinuously(Duration.ofSeconds(5))
                .build();

        DataStream<String> ddlStream = env.fromSource(
                ddlSource,
                WatermarkStrategy.noWatermarks(),
                "DDL Source"
        );

        // DDL专用Sink配置
        DorisSink.Builder<String> ddlSinkBuilder = DorisSink.builder();
        ddlSinkBuilder.setDorisReadOptions(DorisReadOptions.builder().build())
                .setDorisExecutionOptions(DorisExecutionOptions.builder()
                        .setLabelPrefix("ddl-" + UUID.randomUUID())
                        .setStreamLoadProp(new Properties())
                        .build())
                .setDorisOptions(dorisOptions)
                .setSerializer(
                        JsonDebeziumSchemaSerializer.builder()
                                .setDorisOptions(dorisOptions)
                                .setNewSchemaChange(true)
                                .setSchemaChangeMode("sql_parser")  // 推荐使用SQL Parser
                                .build());

        ddlStream.sinkTo(ddlSinkBuilder.build()).name("DDL Processor");

        // ========== 任务2：处理数据变更事件 ==========
        FileSource<String> dataSource = FileSource
                .forRecordStreamFormat(new TextLineInputFormat(), 
                                      new Path("D:/data/mysql_cdc/data/"))
                .monitorContinuously(Duration.ofSeconds(5))
                .build();

        DataStream<String> dataStream = env.fromSource(
                dataSource,
                WatermarkStrategy.noWatermarks(),
                "Data Source"
        );

        // 数据专用Sink配置
        Properties props = new Properties();
        props.setProperty("format", "json");
        props.setProperty("read_json_by_line", "true");

        DorisSink.Builder<String> dataSinkBuilder = DorisSink.builder();
        dataSinkBuilder.setDorisReadOptions(DorisReadOptions.builder().build())
                .setDorisExecutionOptions(DorisExecutionOptions.builder()
                        .setLabelPrefix("data-" + UUID.randomUUID())
                        .setStreamLoadProp(props)
                        .setDeletable(true)
                        .build())
                .setDorisOptions(dorisOptions)
                .setSerializer(
                        JsonDebeziumSchemaSerializer.builder()
                                .setDorisOptions(dorisOptions)
                                .setNewSchemaChange(true)
                                .build());

        dataStream.sinkTo(dataSinkBuilder.build()).name("Data Processor");

        env.execute("Separated Files to Doris");
    }
}

B.3 方案优势与劣势

优势：

• 支持自动建表和schema演化
• 事件处理逻辑清晰
• 可以优先处理DDL保证表存在

劣势：

• 需要维护两套文件流
• 时序可能出现问题（数据到了但表还没创建）
• 复杂度较高

---

方案C：预建表 + 纯数据同步

如果schema相对稳定，可以预先创建表。

C.1 实现步骤

步骤1：一次性建表

public class InitialTableCreator {
    public static void main(String[] args) throws Exception {
        // 使用DatabaseSync的create-table-only模式
        Map<String, String> config = new HashMap<>();
        config.put("mysql-conf.hostname", "localhost");
        config.put("mysql-conf.port", "3307");  // 通过隧道临时连接
        config.put("mysql-conf.username", "root");
        config.put("mysql-conf.password", "123456");
        config.put("mysql-conf.database-name", "test");
        
        config.put("sink.fenodes", "127.0.0.1:8030");
        config.put("sink.username", "root");
        config.put("sink.password", "123456");
        
        // 关键：只建表不同步数据
        config.put("create-table-only", "true");
        
        Configuration configuration = Configuration.fromMap(config);
        MysqlDatabaseSync sync = new MysqlDatabaseSync();
        sync.setConfig(configuration);
        sync.create();
        sync.build();
    }
}

步骤2：修改CDC为纯数据模式

// 生成文件时：设置 includeSchemaChanges(false)
MySqlSource<String> mySqlSource =
        MySqlSource.<String>builder()
                // ... 其他配置
                .includeSchemaChanges(false)  // 不包含DDL
                .build();

步骤3：使用简化的Doris Sink

使用 JsonDebeziumDataChange 仅处理数据变更：

C.2 适用场景

• Schema变更不频繁

• 可接受手动DDL变更

• 追求方案简单性

• 不支持：自动schema演化

---

方案D：Kafka中转

使用Kafka替代文件传输。

D.1 架构

graph LR
    A[外网MySQL CDC] --> B[Kafka<br/>外网]
    B --> C[镜像同步] 
    C --> D[Kafka<br/>内网]
    D --> E[Flink消费] 
    E --> F[Doris]

Loading

D.2 优势

• 天然支持事件顺序
• 支持断点续传
• 高可用性
• 实时性更好

配置示例：

// 外网写入Kafka
FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<>(
    "mysql-cdc-topic",
    new SimpleStringSchema(),
    props
);
cdcStream.addSink(kafkaProducer);

// 内网从Kafka消费
FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(
    "mysql-cdc-topic",
    new SimpleStringSchema(),
    props
);
DataStream<String> stream = env.addSource(kafkaConsumer);

---

三、关键配置说明

3.1 Schema Change模式选择

推荐使用 SQL_PARSER 模式：

• 直接解析DDL语句
• 支持更复杂的DDL操作
• 准确性更高

3.2 关键参数配置

DorisExecutionOptions executionOptions = DorisExecutionOptions.builder()
    .setLabelPrefix("doris-" + UUID.randomUUID())
    .setStreamLoadProp(props)
    .setDeletable(true)                    // 支持DELETE操作
    .setIgnoreUpdateBefore(true)           // 忽略UPDATE的before值
    .setBatchSize(10000)                   // 批次大小
    .setBatchIntervalMs(10000)             // 批次间隔
    .setMaxRetries(3)                      // 重试次数
    .build();

---

四、生产环境最佳实践

4.1 监控告警

// 添加自定义Metrics
env.getConfig().setLatencyTrackingInterval(1000L);

// 监控指标
- 处理延迟
- 失败重试次数
- Schema change成功率
- 数据写入TPS

4.2 容错配置

env.enableCheckpointing(60000);  // 1分钟checkpoint
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);
env.getCheckpointConfig().setCheckpointTimeout(600000);
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
env.getCheckpointConfig().enableExternalizedCheckpoints(
    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
);

4.3 表设计建议

-- Doris建表时使用Unique Key模型（推荐）
CREATE TABLE test.students (
    student_id INT,
    first_name VARCHAR(100),
    last_name VARCHAR(100),
    ...
)
UNIQUE KEY(student_id)
DISTRIBUTED BY HASH(student_id) BUCKETS 10;

所以我认为您应该想实现Schema Change处理,如果必须使用您的架构，需要实现方案B（事件分离），将DDL和DML事件分开处理，确保DDL先执行

配置关键点：

• 使用 includeSchemaChanges(true) 才能捕获DDL
• 设置 schema-change-mode 为 sql_parser 获得最佳兼容性
• 必须保证DDL事件在相应数据变更事件之前处理

[kaori-seasons]

Hi @ladebangbangde , would you be willing to handle this issue?