在Flink中使用CDC同步Oracle中的RowID

Flink CDC(Change Data Capture)是一种用于捕获数据库中数据变化的技术。要在Flink中使用CDC同步Oracle中的RowID，您需要按照以下步骤操作：

1. 确保您的Oracle数据库已经启用了归档日志功能。这可以通过修改参数db_recovery_file_dest_size和db_recovery_file_dest来实现。例如，将db_recovery_file_dest_size设置为10GB，并将db_recovery_file_dest指向一个空闲磁盘空间较大的文件夹。

2. 创建一个具有足够权限的Oracle用户，以便Flink可以访问数据库。这个用户应该能够执行DDL(数据定义语言)语句，如ALTER TABLE和CREATE TABLE。

3. 为要同步的表或数据库启用增量日志记录。这通常是通过在ALTER TABLE语句后添加ENABLE ROWID LOGGING来完成的。例如：

   ALTER TABLE my_table ENABLE ROWID LOGGING;
   

4. 安装并配置Apache Flink环境。确保已安装Java JDK和Scala编译器。然后，从Apache Flink官方网站下载最新版本的Flink发行版。

5. 克隆Flink CDC connector for Oracle仓库到本地：

   git clone https://github.com/ververica/flink-cdc-connectors.git
   

6. 将flink-connector-oracle-cdc模块导入到您的项目中。这可以通过将该模块添加到项目的pom.xml文件中来完成。例如：

   <dependency>
       <groupId>org.apache.flink</groupId>
       <artifactId>flink-connector-oracle-cdc</artifactId>
       <version>{
       {FLINK_VERSION}}</version>
   </dependency>
   

7. 在Flink作业中添加一个新的DataStream应用程序，并在其中添加一个ReadOnlyTableSourceFunction函数。此函数将读取来自Oracle数据库的数据。例如：

   val tableSource = new ReadOnlyTableSourceFunction("jdbc:oracle:thin:@localhost:1521:xe", "my_user", "my_password", "my_schema.my_table")
   

8. 运行Flink作业以启动数据流处理任务。这可以通过在命令行中输入以下命令来完成：

   ```
   ./bin/flink run -m "exec" -c org.apache.flink.streaming.api.scala.StreamExecutionEnvironment \
   -p output_path output_path \
   -p checkpoint_interval 1000 \
   -p parallelism 1 \
   -p table_source tableSource \
   -p job_name MyJob \
   -p zookeeper_quorum localhost:2181 \
   -p group_id testGroup \
   -p application_timeout 60 \
   -p rest_port 8081 \
   -p rest_addresses localhost:8081 \
   -p state_backend_type rocksdb \
   -p state_backend_path file:///opt/cloudera/parcels/CDH-5.13.-SNAPSHOT/etc/hadoop/conf \
   -p keyed_state_backend_path file:///opt/cloudera/parcels/CDH-5.13.-SNAPSHOT/etc/hadoop/conf \
   -p default_parallelism 1 \
   -p taskmanager_memory 4096 \
   -p network_card_memory --- \
   -p slot_num 1 \
   -p slots_per_task 1 \
   -p yarn_session_application_timeout 60 \
   -p hadoop_binary_home /usr/lib/hadoop/ \
   -p hive_metastore_uris thrift://localhost:9083 \
   -p oozie_url http://localhost:11