这是使用Flink Table API在DataStreams上执行FULL OUTER JOIN的完整工作示例。有关Table API下的
DataStream API集成
的更多详细信息,请参阅Flink官方
页面。
步骤
1. Convert both dataStreams into tables.
2. Registers tables as views to execute SQL query.
3. Execute full outer join SQL query on registered views.
4. Result of the SQL query will be a table.
5. Convert the result table to dataStream using toDataStream.
public class DataStreamFullOuterJoinUsingTable {
public static void main(String[] args) throws Exception {
final StreamExecutionEnvironment streamEnv =
StreamExecutionEnvironment.getExecutionEnvironment();
streamEnv.setRuntimeMode(RuntimeExecutionMode.BATCH);
final StreamTableEnvironment tableEnv = StreamTableEnvironment.create(streamEnv);
DataStream<String> nameStream = streamEnv.fromElements("Alice", "Bob", "John");
Table nameTable = tableEnv.fromDataStream(nameStream).as("name");
tableEnv.createTemporaryView("nameTable", nameTable);
DataStream<String> detailStream = streamEnv.fromElements("Mike", "Sam", "Adam", "Alice");
Table detailTable = tableEnv.fromDataStream(detailStream).as("detail");
tableEnv.createTemporaryView("detailTable", detailTable);
Table result =
tableEnv.sqlQuery(
"SELECT * FROM "
+ "nameTable FULL OUTER JOIN detailTable "
+ "ON nameTable.name = detailTable.detail");
DataStream<String> resultStream =
tableEnv
.toDataStream(result)
.map(
(MapFunction<Row, String>)
row -> "(" + row.getField(0) + "), (" + row.getField(1) + ")");
resultStream.print();
streamEnv.execute();
}
}
除了 flink-streaming-java 之外,您还需要以下 3 个依赖项(提供的范围)来使 DataStream 和 Table 能够一起工作。
1. flink-table-api-java
2. flink-table-api-java-bridge
3. flink-table-planner_2.12
一旦你知道如何将数据流转换为表格,执行一个SQL查询,然后再将其转换回数据流,你就可以解决许多类似的问题。
connect
实现这个是否可能? - MJeremy