如何在Flink中使用表API消费两个源码流

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.types.DataType;

TableSchema schema = TableSchema.builder()
    .field("id", DataTypes.STRING())
    .field("value", DataTypes.INT())
    .build();

Table sourceTable1 = tEnv.fromDataStream(stream1, schema);
tEnv.createTemporaryView("sourceTable1", sourceTable1);

Table sourceTable2 = tEnv.fromDataStream(stream2, schema);
tEnv.createTemporaryView("sourceTable2", sourceTable2);

Table resultTable = tEnv.sqlQuery("SELECT * FROM sourceTable1 JOIN sourceTable2 ON sourceTable1.id = sourceTable2.id");

DataStream<Row> resultStream = tEnv.toAppendStream(resultTable, Types.ROW(Types.STRING, Types.INT, Types.STRING, Types.INT));
resultStream.print();