Flink CDC我5张oracle表做关联，怎么用区间join？

在 Flink CDC 中，可以使用区间连接（interval join）来对多个 Oracle 表进行关联。区间连接是指在两个表之间进行关联操作，基于一个范围条件来匹配符合条件的数据。

下面是一个示例，演示了如何在 Flink CDC 中使用区间连接：
```StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1); // 设置并行度

Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "localhost:9092");
properties.setProperty("group.id", "cdc_group");
properties.setProperty("database.server.name", "myOracleServer");

FlinkCDCConsumer consumer = new FlinkCDCConsumer<>("oracle", DebeziumDeserializationSchema.forString(), properties);

DataStream stream = env.addSource(consumer);

// 创建第一个表的流
DataStream> table1Stream = stream
.filter(record -> record.contains("table1")) // 根据 CDC 数据中的表名过滤出需要的表数据
.map(record -> {
// 解析 CDC 数据中的字段，生成 Tuple2
// 假设 CDC 数据中表1的字段为
String[] fields = record.split(",");
String key = fields[0];
int value = Integer.parseInt(fields[1]);
return Tuple2.of(key, value);
});

// 创建第二个表的流
DataStream> table2Stream = stream
.filter(record -> record.contains("table2")) // 根据 CDC 数据中的表名过滤出需要的表数据
.map(record -> {
// 解析 CDC 数据中的字段，生成 Tuple3
// 假设 CDC 数据中表2的字段为
String[] fields = record.split(",");
String key = fields[0];
int start = Integer.parseInt(fields[1]);
int end = Integer.parseInt(fields[2]);
return Tuple3.of(key, start, end);
});

// 执行区间连接操作
DataStream> resultStream = table1Stream
.keyBy(0) // 按照表1的 key 字段进行分组
.intervalJoin(table2Stream.keyBy(0)) // 按照表2的 key 字段进行分组
.between(Time.seconds(0), Time.seconds(10)) // 设置区间范围
.process(new IntervalJoinFunction, Tuple3, Tuple3>() {
@Override
public void processElement(Tuple2 left, Tuple3 right, Context ctx,

```