Flink 的 Table API和 SQL是流批统一的 API。 这意味着 Table API & SQL 在无论有限的批式输入还是无限的流式输入下，都具有相同的语义。 因为传统的关系代数以及 SQL 最开始都是为了批式处理而设计的， 关系型查询在流式场景下不如在批式场景下容易理解.
动态表(Dynamic Tables)是 Flink 的支持流数据的 Table API 和 SQL 的核心概念。
与表示批处理数据的静态表不同，动态表是 随时间变化的。可以像查询静态批处理表一样查询它们。查询动态表将生成一个 连续查询(Continuous Query)。一个连续查询永远不会终止，结果会生成一个动态表。查询不断更新其(动态)结果表，以反映其(动态)输入表上的更改。
TableAPI
首先需要导入依赖
<dependency>
<groupId>org.apache.flinkgroupId>
<artifactId>flink-table-planner-blink_${scala.binary.version}artifactId>
<version>${flink.version}version>
<scope>providedscope>
dependency>
<dependency>
<groupId>org.apache.flinkgroupId>
<artifactId>flink-streaming-scala_${scala.binary.version}artifactId>
<version>${flink.version}version>
<scope>providedscope>
dependency>
<dependency>
<groupId>org.apache.flinkgroupId>
<artifactId>flink-csvartifactId>
<version>${flink.version}version>
dependency>
<dependency>
<groupId>org.apache.flinkgroupId>
<artifactId>flink-jsonartifactId>
<version>${flink.version}version>
dependency>
这里需要注意的问题：
1.TableAPI 中将动态表转换为流时有两种方法
DataStream<Row> rowDataStream = tableEnvironment.toAppendStream(result, Row.class);
toAppendStream方法只能在查询时使用，不能使用包含聚合函数等更新语句
DataStream<Tuple2<Boolean, Row>> tuple2DataStream = tableEnvironment.toRetractStream(select, Row.class);
toRetractStream则可以使用
2.上述两种方法内传入的参数Row.class，表示将表中查询出的数据封装为行类型，也就是对每行进行封装，解决查询出的数据列少于或者多于原表。如何能够确保所查询的数据与之前封装的Bean有完全一致的结构则也可以封装为原Bean.class
代码实现：
package net.cyan.FlinkSql; import net.cyan.POJO.WaterSensor; import org.apache.flink.api.common.functions.FilterFunction; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.configuration.Configuration; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.table.api.Table; import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; import org.apache.flink.types.Row; import static org.apache.flink.table.api.Expressions.$;
使用TableAPI读取文件时，我们首先需要知道去哪里读取也就是文件路径、读取文件的格式、读取出来的数据的结构也就是结果表的表结构及表名
package net.cyan.FlinkSql; import org.apache.flink.configuration.Configuration; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.table.api.DataTypes; import org.apache.flink.table.api.Table; import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; import org.apache.flink.table.descriptors.Csv; import org.apache.flink.table.descriptors.FileSystem; import org.apache.flink.table.descriptors.Schema; import org.apache.flink.table.types.DataType; import static org.apache.flink.table.api.Expressions.$; public class Demo2_readWriteText { public static void main(String[] args) {
基本流程
1>需要创建表的运行环境
2>创建查询出的数据写出结构
3> 创建kafka连接
4> 进行查询
5> 创建写入kafka连接
6> 写入
完整代码如下
package net.cyan.FlinkSql; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.table.api.DataTypes; import org.apache.flink.table.api.Table; import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; import org.apache.flink.table.descriptors.Json; import org.apache.flink.table.descriptors.Kafka; import org.apache.flink.table.descriptors.Schema; public class Demo5_readWriteKafka { public static void main(String[] args) {
Original: https://www.cnblogs.com/CYan521/p/16845742.html
Author: 再美不及姑娘你
Title: FlinkSql之TableAPI详解
原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/713564/
转载文章受原作者版权保护。转载请注明原作者出处！