flink cdc 同步oracle 19c 数据库，是2.4.0吗？

Flink CDC 同步 Oracle 19c 数据库的配置与说明

1. Flink CDC 版本支持

根据知识库信息，flink-connector-oracle-cdc 的版本为 2.4.0 是可以支持同步 Oracle 19c 数据库的。但需要注意以下几点： - CDC 连接器的版本与 Flink 引擎版本的对应关系：确保使用的 flink-connector-oracle-cdc 版本与 Flink 引擎版本兼容。例如，VVR（Ververica Runtime）版本与社区版 CDC 的对应关系需要匹配。 - 如果您使用的是阿里云实时计算 Flink 版，建议优先使用内置的 Oracle CDC 连接器，而不是社区版连接器，以避免兼容性问题。

2. Oracle 数据库配置

在同步 Oracle 19c 数据库之前，需要对数据库进行必要的配置： - 启用归档日志模式：确保 Oracle 数据库启用了归档日志模式（Archivelog Mode），这是 CDC 同步的基础。 - 启用补充日志记录：

ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA (PRIMARY KEY) COLUMNS;

补充日志记录是 Oracle CDC 连接器捕获数据变更的前提条件。 - 创建高权限用户：确保用于同步的 Oracle 用户具有以下权限： - SELECT ANY TRANSACTION - LOGMINING - SELECT_CATALOG_ROLE - 对目标表的 SELECT 权限 - 配置 LogMiner：确保 LogMiner 已正确配置，并且能够读取归档日志和在线重做日志。

3. Flink 作业配置

以下是使用 flink-connector-oracle-cdc 2.4.0 同步 Oracle 19c 数据库的配置步骤：

3.1 上传连接器 JAR 包

• 如果使用的是社区版连接器，请从 Apache Flink CDC 官方页面 下载对应版本的 JAR 包。
• 在阿里云实时计算控制台中，上传 JAR 包并创建自定义连接器。

3.2 编写 Flink SQL

以下是一个示例 SQL 配置，用于同步 Oracle 19c 数据库中的表到目标存储（如 SelectDB）：

CREATE TABLE oracle_source (
    id INT,
    name STRING,
    update_time TIMESTAMP(3),
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'oracle-cdc',
    'hostname' = '<your-oracle-host>',
    'port' = '1521',
    'username' = '<your-oracle-username>',
    'password' = '<your-oracle-password>',
    'database-name' = '<your-database-name>',
    'schema-name' = '<your-schema-name>',
    'table-name' = '<your-table-name>',
    'debezium.snapshot.mode' = 'initial'
);

CREATE TABLE selectdb_sink (
    id INT,
    name STRING,
    update_time TIMESTAMP(3),
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'selectdb',
    'fenodes' = '<your-selectdb-fenodes>',
    'username' = '<your-selectdb-username>',
    'password' = '<your-selectdb-password>',
    'table.identifier' = '<your-target-table>'
);

INSERT INTO selectdb_sink
SELECT * FROM oracle_source;

3.3 参数说明

• Oracle CDC 参数：
  • hostname：Oracle 数据库的主机地址。
  • port：Oracle 数据库的监听端口，默认为 1521。
  • username 和 password：具有必要权限的 Oracle 用户名和密码。
  • database-name：Oracle 数据库名称。
  • schema-name：目标表所在的 Schema 名称。
  • table-name：需要同步的目标表名称。
  • debezium.snapshot.mode：快照模式，推荐使用 initial，表示首次全量同步后切换为增量同步。
• SelectDB Sink 参数：
  • fenodes：SelectDB 的前端节点地址。
  • username 和 password：SelectDB 的用户名和密码。
  • table.identifier：目标表的标识符。

4. 运行 Flink 作业

通过 Flink CLI 或阿里云实时计算控制台提交作业。以下是一个示例命令：

<FLINK_HOME>/bin/flink run \
-Dexecution.checkpointing.interval=10s \
-Dparallelism.default=1 \
-c org.apache.doris.flink.tools.cdc.CdcTools \
lib/flink-doris-connector-1.16-1.5.2.jar \
oracle-sync-database \
-database test_db \
-oracle-conf hostname=127.0.0.1 \
-oracle-conf port=1521 \
-oracle-conf username=admin \
-oracle-conf password="password" \
-oracle-conf database-name=XE \
-oracle-conf schema-name=ADMIN \
-including-tables "tbl1|test.*" \
-sink-conf fenodes=selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080 \
-sink-conf username=admin \
-sink-conf password=****

5. 注意事项

• 性能优化：建议调整 Flink 的 checkpoint 间隔和并行度，以适应数据同步的需求。
• 错误处理：如果同步过程中出现错误，请检查 Oracle 数据库的日志配置以及 Flink 作业的连接参数。
• 版本兼容性：确保 flink-connector-oracle-cdc 的版本与 Flink 引擎版本匹配，避免因版本不一致导致的问题。

通过以上步骤，您可以成功使用 flink-connector-oracle-cdc 2.4.0 实现 Oracle 19c 数据库的数据同步。