Flink CDC里这个是Kafka写入Doris，Doris表一条数据也没有是为啥？

根据您提供的问题描述，Flink CDC 写入 Doris 时，Doris 表中没有数据，并且日志中出现了 ddl can not do schema change 的错误信息。以下是针对该问题的详细分析和解决思路：

1. 问题背景与可能原因

从错误信息 ddl can not do schema change 可以推测，问题可能与以下因素相关： - Schema 不匹配：Flink CDC 数据流中的 Schema 与 Doris 表的 Schema 不一致，导致写入失败。 - DDL 操作限制：Doris 表可能不支持动态 Schema 变更，而 Flink CDC 数据流中包含了 DDL 操作（如表结构变更）。 - 数据格式问题：Flink CDC 数据流中的 JSON 格式可能未正确解析，导致 Doris 无法识别数据。

结合知识库资料，Flink Doris Connector 在将数据写入 Doris 时，需要确保数据源、转换逻辑和目标表的 Schema 完全匹配。如果 Schema 不匹配或存在动态变更，可能会导致写入失败。

2. 排查步骤

2.1 检查 Doris 表的 Schema

• 确认 Doris 表的 Schema 是否与 Flink CDC 数据流中的字段完全一致。
• 如果 Doris 表的字段类型或名称与 Flink CDC 数据流不一致，需要调整 Doris 表的定义，使其与数据流匹配。

2.2 检查 Flink CDC 数据流

• 使用 Flink 的调试工具或日志，查看 Flink CDC 数据流的实际内容，确认是否包含 DDL 操作。
• 如果数据流中包含 DDL 操作（如 ALTER TABLE），需要在 Flink 作业中过滤掉这些操作，仅保留 DML 操作（如 INSERT、UPDATE、DELETE）。

2.3 检查 Flink Doris Connector 配置

• 确认 Flink Doris Connector 的配置是否正确，特别是以下参数：
  • sink.properties.format：确保数据格式为 JSON 或其他 Doris 支持的格式。
  • sink.properties.strip_outer_array：如果数据流是数组格式，需要设置为 true。
  • sink.properties.schema-change：如果 Doris 表支持 Schema 变更，可以尝试设置为 true。

2.4 检查 Doris 表的写入权限

• 确保 Flink 作业运行的用户对 Doris 表具有写入权限。
• 如果 Doris 表启用了白名单机制，需将 Flink 作业的 IP 地址添加到白名单中。

3. 解决方案

3.1 过滤 DDL 操作

如果 Flink CDC 数据流中包含 DDL 操作，可以通过以下方式过滤掉： - 在 Flink SQL 中使用 WHERE 子句过滤掉非 DML 操作的数据。 - 示例：

CREATE TABLE kafka_source (
    data STRING,
    op STRING
) WITH (
    'connector' = 'kafka',
    'topic' = 'your_topic',
    'properties.bootstrap.servers' = 'your_kafka_broker',
    'format' = 'json'
);

CREATE TABLE doris_sink (
    id INT,
    name STRING
) WITH (
    'connector' = 'doris',
    'fenodes' = 'your_doris_fenodes',
    'table.identifier' = 'your_table',
    'username' = 'your_username',
    'password' = 'your_password'
);

INSERT INTO doris_sink
SELECT 
    CAST(data['id'] AS INT) AS id,
    data['name'] AS name
FROM kafka_source
WHERE op = 'INSERT' OR op = 'UPDATE';

3.2 调整 Doris 表的 Schema

• 如果 Doris 表的 Schema 与 Flink CDC 数据流不一致，可以通过以下方式调整：
  • 修改 Doris 表的字段类型或名称，使其与数据流一致。
  • 如果 Doris 表不支持动态 Schema 变更，可以在 Flink 作业中进行字段映射和转换。

3.3 启用 Schema 变更支持

• 如果 Doris 表支持 Schema 变更，可以在 Flink Doris Connector 中启用 schema-change 参数：

  sink.properties.schema-change=true
  

• 注意：启用此参数后，Doris 表会尝试自动适配数据流中的 Schema 变更，但可能会导致性能下降。

4. 常见问题与注意事项

4.1 数据格式问题

• 如果 Flink CDC 数据流中的 JSON 格式不符合 Doris 的要求，可能会导致写入失败。可以通过以下方式解决：
  • 在 Flink 作业中对 JSON 数据进行预处理，确保其符合 Doris 的格式要求。
  • 示例：

  DataStream<String> processedStream = rawStream.map(json -> {
      // 对 JSON 数据进行预处理
      return transformJson(json);
  });
  

4.2 网络连通性

• 确保 Flink 和 Doris 之间的网络是连通的。如果 Flink 和 Doris 位于不同的 VPC 中，需要配置网络互通。

4.3 日志分析

• 如果问题仍未解决，可以通过以下方式进一步分析日志：
  • 查看 Flink 作业的日志，确认是否有其他异常信息。
  • 查看 Doris 的日志，确认是否有写入失败的记录。

5. 总结

通过以上排查和解决方案，您可以逐步定位并解决 Flink CDC 写入 Doris 时的问题。重点在于： - Schema 匹配：确保 Flink CDC 数据流与 Doris 表的 Schema 一致。 - 过滤 DDL 操作：避免 DDL 操作导致写入失败。 - 配置优化：调整 Flink Doris Connector 的参数，确保数据格式和写入逻辑正确。

如果问题仍然存在，建议提供更详细的日志信息以便进一步分析。