Flink CDC，data-stream 如何实现想 sql-join 一样的全量 join 呢？

Flink CDC 的 data-stream API 中没有直接支持全量 join 的算子。但是，你可以使用以下方法实现类似于 SQL 全量 join 的功能：

方法 1：使用侧输出

创建两个流，分别代表两个要连接的表。
使用 connect() 算子将这两个流连接起来。
在连接流上使用 process() 算子，在其中执行全量 join 逻辑。
使用 output() 方法将连接结果输出到侧输出。
从主输出中获取要连接的数据，并使用 process() 算子将侧输出中的连接结果与主输出中的数据合并。
代码示例：

DataStream stream1 = ...;
DataStream stream2 = ...;

// 连接两个流
ConnectedStreams connectedStreams = stream1.connect(stream2);

// 在连接流上执行全量 join 逻辑
SingleOutputStreamOperator joinedStream = connectedStreams
.process(new ProcessFunction() {
@Override
public void processElement(Row value, Context ctx, Collector out) throws Exception {
// 获取另一个流中的数据
Row otherRow = ctx.getInput(1).next();

        // 执行全量 join 逻辑
        Row joinedRow = join(value, otherRow);

        // 输出连接结果
        out.collect(joinedRow);
    }
});

// 从主输出中获取要连接的数据
DataStream mainStream = connectedStreams.getFirstInput();

// 合并侧输出中的连接结果与主输出中的数据
DataStream resultStream = mainStream
.process(new ProcessFunction() {
@Override
public void processElement(Row value, Context ctx, Collector out) throws Exception {
// 获取侧输出中的连接结果
Row joinedRow = ctx.getSideInput(joinedStream).next();

        // 合并数据
        Row mergedRow = merge(value, joinedRow);

        // 输出合并后的数据
        out.collect(mergedRow);
    }
})
.setSideInput(joinedStream);

方法 2：使用状态后端

创建两个流，分别代表两个要连接的表。
使用 broadcast() 算子将一个流（较小的流）广播到所有并行实例。
在广播流上使用 map() 算子，将数据存储在状态后端。
在另一个流（较大的流）上使用 process() 算子，与状态后端中的数据进行全量 join。
代码示例：

DataStream stream1 = ...;
DataStream stream2 = ...;

// 广播较小的流
DataStream broadcastStream = stream1.broadcast();

// 将广播流中的数据存储在状态后端
DataStream updateStateStream = broadcastStream
.map(new MapFunction() {
@Override
public Void map(Row value) throws Exception {
// 更新状态后端
backend.update(value);
return null;
}
});

// 在较大的流上执行全量 join 逻辑
DataStream joinedStream = stream2
.process(new ProcessFunction() {
@Override
public void processElement(Row value, Context ctx, Collector out) throws Exception {
// 从状态后端获取数据
Iterable otherRows = backend.get(value);

        // 执行全量 join 逻辑
        for (Row otherRow : otherRows) {
            Row joinedRow = join(value, otherRow);
            out.collect(joinedRow);
        }
    }
})
.setMaxParallelism(1); // 将算子设置为单并行度，以确保所有数据都存储在同一个状态后端实例中

// 启动状态后端
updateStateStream.getExecutionEnvironment().execute();
这两种方法都可以实现类似于 SQL 全量 join 的功能，但具体选择哪种方法取决于数据量和性能要求。