Flink CDC 里有用datastrea来cdc oracle数据的吗？

是的，Flink CDC 支持使用 DataStream API 来捕获 Oracle 数据库的 CDC 数据。通过使用 Flink CDC connector，可以实现对 Oracle 数据库的实时增量数据同步。

Flink CDC 使用 DataStream API 进行 Oracle 数据的 CDC

单元1：概述

Apache Flink 是一个开源的流处理框架，它提供了丰富的 API 来处理实时数据，Flink CDC（Change Data Capture）是 Flink 提供的一种用于捕获数据库中的数据变更的工具，通过 Flink CDC，我们可以实时地捕获到数据库中的数据变更，然后对这些变更进行处理，在这篇文章中，我们将介绍如何使用 Flink CDC 的 DataStream API 来捕获 Oracle 数据库中的数据变更。

单元2：准备工作

在使用 Flink CDC 之前，我们需要先完成以下准备工作：

1、安装并配置 Flink：请参考 Flink 官方文档（https://flink.apache.org/）来完成 Flink 的安装和配置。

2、添加 Flink CDC 依赖：在项目的 pom.xml 文件中添加 Flink CDC 的依赖。

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flinkconnectordebezium_2.11</artifactId>
    <version>${flink.version}</version>
</dependency>

3、准备 Oracle 数据库：确保已经安装了 Oracle 数据库，并且已经创建了需要监控的表和模式，需要在 Oracle 数据库中启用 CDC，具体操作可以参考 Oracle 官方文档（https://docs.oracle.com/en/database/oracle/oracledatabase/19/cncpt/changedatacaptureconcepts.html）。

单元3：使用 DataStream API 捕获 Oracle 数据变更

接下来，我们将使用 Flink CDC 的 DataStream API 来捕获 Oracle 数据库中的数据变更，以下是一个简单的示例：

1、创建 Flink 执行环境：

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.connector.debezium.DebeziumSourceFunction;
import org.apache.flink.connector.debezium.config.DebeziumDeserializationSchema;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcConnectionOptionsBuilder;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.catalog.debezium.*;
import org.apache.flink.table.catalog.debezium.DebeziumOptionsBuilder;
import org.apache.flink.table.catalog.debezium.DebeziumCatalog;
import org.apache.flink.table.descriptors.*;
import org.__subproject__.__subpackage____subclass__\__submethod___; // 根据实际项目情况替换这部分代码

2、设置 Flink SQL：

String sourceDDL = "CREATE TABLE source_table (...)"; // 根据实际表结构替换这部分代码
env().executeSql(sourceDDL);

3、注册 Flink Table：

DebeziumCatalog debeziumCatalog = new MyDebeziumCatalog(); // 根据实际项目情况替换这部分代码
String databaseName = "my_database"; // 根据实际数据库名称替换这部分代码
String tableName = "my_table"; // 根据实际表名称替换这部分代码
debeziumCatalog.createDatabase(databaseName, true, false, false); // 根据实际需求设置参数
debeziumCatalog.createTable(databaseName, tableName, sourceDDL, true); // 根据实际需求设置参数

4、使用 DataStream API 捕获数据变更：

String url = "jdbc:oracle:thin:@localhost:1521:mydb"; // 根据实际数据库地址替换这部分代码
String user = "username"; // 根据实际用户名替换这部分代码
String password = "password"; // 根据实际密码替换这部分代码
String dbTable = "my_table"; // 根据实际表名称替换这部分代码
DebeziumSourceFunction<Row> sourceFunction = DeserializationSchemaUtils // 根据实际项目情况替换这部分代码；
DataStream<Row> rowDataStream = env().addSource(new RowDataSourceFunctionAdapter<>(sourceFunction)); // 根据实际项目情况替换这部分代码；

5、对捕获到的数据进行处理：在这一步，我们可以对接收到的数据进行各种处理，例如过滤、聚合等，这里我们简单地将数据打印出来：

rowDataStream.print(); // 根据实际需求设置参数

6、启动 Flink 作业：我们启动 Flink 作业来捕获数据变更，注意，这里的 env()、sourceFunction、rowDataStream、sourceDDL、databaseName、tableName、url、user、password、dbTable、debeziumCatalog 等变量都需要根据实际项目情况进行替换。