Table API 标准结构

EXP

下面是一个标准的Table API 的例子

public class FlinkTableStandardStructure {
    public static void main(String[] args) {
        //1、创建TableEnvironment
        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                .inStreamingMode()//默认就是StreamingMode
                //.inBatchMode()
                .build();
        TableEnvironment tEnv = TableEnvironment.create(settings);
        //2、创建source table: 1）读取外部表；2）从Table API或者SQL查询结果创建表
        Table projTable = tEnv.fromValues(
                DataTypes.ROW(
                        DataTypes.FIELD("id", DataTypes.DECIMAL(10, 2)),
                        DataTypes.FIELD("name", DataTypes.STRING())
                ),
                row(1, "zhangsan"),
                row(2L, "lisi")
        ).select($("id"), $("name"));
        //注册表到catalog(可选的)
        tEnv.createTemporaryView("sourceTable", projTable);
        //3、创建sink table
        final Schema schema = Schema.newBuilder()
                .column("id", DataTypes.DECIMAL(10, 2))
                .column("name", DataTypes.STRING())
                .build();
        tEnv.createTemporaryTable("sinkTable", TableDescriptor.forConnector("print")
                .schema(schema)
                .build());
        //4、Table API执行查询(可以执行多次查询，中间表可以注册到catalog也可以不注册)
        Table resultTable = tEnv.from("sourceTable").select($("id"), $("name"));
//        Table resultTable = projTable.select($("id"), $("name"));
        //5、输出(包括执行,不需要单独在调用tEnv.execute("job"))
        resultTable.executeInsert("sinkTable");
    }
}

两种TableEnvironment

1、TableEnvironment

标准的做法是创建TableEnvironment，在创建 TableEnv 的时候，可以传⼊⼀个 EnvironmentSettings 或者 TableConfig 参数，⽤来配置 TableEnvironment 的⼀些特性：注意：Flink1.14开始删除了其他的执⾏器了，只保留了BlinkPlanner

功能：

注册catalog(可以理解为数据系统实例，⽐如某个HBase集群，某个MySQL服务器)
在catalog注册库和表
加载插件模块
执⾏SQL查询
注册UDF
DataStream和Table互转(仅仅在StreamExecutionEnvironment下)

EnvironmentSettings settings = EnvironmentSettings
 .newInstance()
 .useBlinkPlanner()//Flink1.14开始就删除了其他的执⾏器了，只保留了BlinkPlanner
 .inStreamingMode()//默认就是StreamingMode
 //.inBatchMode()
 .build();
TableEnvironment tEnv = TableEnvironment.create(settings);

2、StreamTableEnvironment

如果要混⽤DataStream和Table API/SQL，可以使⽤StreamTableEnvironment：

//1、获取Stream执⾏环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//2、创建表执⾏环境
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

TableEnvironment和StreamExecutionEnvironment⼆选⼀即可。

TableEnviromnet的配置

// instantiate table environment
TableEnvironment tEnv = ...
// access flink configuration
Configuration configuration = tEnv.getConfig().getConfiguration();
// set low-level key-value options
configuration.setString("table.exec.mini-batch.enabled", "true");
configuration.setString("table.exec.mini-batch.allow-latency", "5 s");
configuration.setString("table.exec.mini-batch.size", "5000");

具体配置项请参考如下链接： https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/table/config/

创建表和注册表

有了TableEnvironment，我们接下来就需要创建表并注册表，注册表是可选的。

创建表的两种⽅式

创建表是为了在表上执⾏TableAPI或者SQL查询，可以通过如下两种⽅式创建表

创建表的方式	说明	代码示例
虚拟表	从现有的table创建表，通常是从Table API&SQL查询结果创建表，这种一般是中间表	tEnv.createTemporaryView("sourceTable", projTable);
常规表	另外⼀种更实⽤的⽅式是通过外部数据创建表，例如⽂件, 数据库表, 或者消息队列，这种⼀般是创建输⼊表或者输出表，采⽤ connector方式创建	tableEnv.createTable("SourceTableA", sourceDescriptor); tableEnv.createTemporaryTable("SourceTableB", sourceDescriptor)

创建虚拟表

Table projTable = tEnv.fromValues(
 DataTypes.ROW(
 DataTypes.FIELD("id", DataTypes.DECIMAL(10, 2)),
 DataTypes.FIELD("name", DataTypes.STRING())
 ),
 row(1, "zhangsan"),
 row(2L, "lisi")
).select($("id"), $("name"));
//createTemporaryView
tEnv.createTemporaryView("sourceTable", projTable);

创建常规表

final TableDescriptor sourceDescriptor = TableDescriptor.forConnector("datagen")
 .schema(Schema.newBuilder()
 .column("f0", DataTypes.STRING())
 .build())
 .option(DataGenOptions.ROWS_PER_SECOND, 100)
 .build();
tableEnv.createTable("SourceTableA", sourceDescriptor);
tableEnv.createTemporaryTable("SourceTableB", sourceDescriptor);
// 也可使⽤SQL DDL
tableEnv.executeSql("CREATE [TEMPORARY] TABLE MyTable (...) WITH (...)")

注意：具体的Connector可以参考如下链接： https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/connectors/table/overview/

临时表和持久表

注册表是为了把计算的中间结果注册为⼀个表，供同⼀Flink session后续访问或者跨Flink session访问。

只有常规的表(TABLES)才分临时表和持久表

表类型	说明
临时表	同一个Flink session可访问，跨Flink session 不能访问
持久表	同一个Flink session和跨Flink session 都能访问

EXP

final TableDescriptor sourceDescriptor = TableDescriptor.forConnector("datagen")
 .schema(Schema.newBuilder()
 .column("f0", DataTypes.STRING())
 .build())
 .option(DataGenOptions.ROWS_PER_SECOND, 100)
 .build();
//持久表
tableEnv.createTable("SourceTableA", sourceDescriptor);
//临时表
tableEnv.createTemporaryTable("SourceTableB", sourceDescriptor)

表的标识符

注册表时可以指定唯⼀标识符，格式为：

Catalog(⽬录).database(数据库).objectname(表名)

如果没有指定Catalog或database，就使⽤当前默认值，即tEnv.useCatalog("...")和tEnv.useDatabase("...")指定的默认值，如果tEnv没有指定，则默认值default_catalog.default_database

EXP

TableEnvironment tEnv = ...;
//当前Flink会话切换到custom_catalog.custom_database
tEnv.useCatalog("custom_catalog");
tEnv.useDatabase("custom_database");
Table table = ...;
//在custom_catalog.custom_database下，把table注册为⼀个名叫exampleView的视图表
tableEnv.createTemporaryView("exampleView", table);
//在custom_catalog.other_database下，把table注册为⼀个名叫exampleView的视图表
tableEnv.createTemporaryView("other_database.exampleView", table);
//在custom_catalog.custom_database下，把table注册为⼀个名叫exampleView.View的视图表,注意转义字
符
tableEnv.createTemporaryView("`example.View`", table);
//other_catalog.other_database，把table注册为⼀个名叫exampleView的视图表
tableEnv.createTemporaryView("other_catalog.other_database.exampleView", table);

Table API 查询

常规查询

/**
 * Flink Table API执行聚合操作
 */
public class FlinkTableAggr {
    public static void main(String[] args) throws Exception{
        //1、获取Stream执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2、创建表执行环境
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        env.setParallelism(1);
        //3、读取数据
        DataStream<ClickLogs> clickLogs = env.fromElements(
                "Mary,./home,2022-02-02 12:00:00",
                "Bob,./cart,2022-02-02 12:00:00",
                "Mary,./prod?id=1,2022-02-02 12:00:05",
                "Liz,./home,2022-02-02 12:01:00",
                "Bob,./prod?id=3,2022-02-02 12:01:30",
                "Mary,./prod?id=7,2022-02-02 12:01:45"
        ).map(event -> {
            String[] props = event.split(",");
            return ClickLogs
                    .builder()
                    .user(props[0])
                    .url(props[1])
                    .cTime(props[2])
                    .build();
        });
        //4、流转换为动态表
        Table table = tEnv.fromDataStream(clickLogs);
        //5、执行Table API查询/SQL查询
        /**
         * select
         *  user,
         *  count(url) as cnt
         * from clicks
         * group by user
         */
        Table resultTable = table
                .groupBy($("user"))
                .aggregate($("url").count().as("cnt"))
                .select($("user"),$("cnt"));
//        System.out.println(resultTable.explain());
        //6、将Table转换为DataStream
        //聚合操作必须是撤回流
        DataStream<Tuple2<Boolean,Row>> selectedClickLogs = tEnv.toRetractStream(resultTable,Row.class);
        //7、处理结果：打印/输出
        selectedClickLogs.print();
        //8、执行
        env.execute("FlinkTableAggr");
    }
}

混⽤Table API和SQL查询

public class MixTableAPIAndSQL {
    public static void main(String[] args) {
        //1、创建TableEnvironment
        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                .build();
        TableEnvironment tEnv = TableEnvironment.create(settings);
        //2、创建source table
        Table projTable = tEnv.fromValues(
                DataTypes.ROW(
                        DataTypes.FIELD("user", DataTypes.STRING()),
                        DataTypes.FIELD("url", DataTypes.STRING()),
                        DataTypes.FIELD("cTime", DataTypes.STRING())
                ),
                row("Mary", "./home","2022-02-02 12:00:00"),
                row("Bob", "./cart","2022-02-02 12:00:00"),
                row("Mary", "./prod?id=1","2022-02-02 12:00:05"),
                row("Liz", "./home","2022-02-02 12:01:00"),
                row("Bob", "./prod?id=3","2022-02-02 12:01:30"),
                row("Mary", "./prod?id=7","2022-02-02 12:01:45")
        ).select($("user"), $("url"),$("cTime"));
        //注册表到catalog(可选的)
        tEnv.createTemporaryView("sourceTable", projTable);
        //3、创建sink table
        final Schema schema = Schema.newBuilder()
                .column("user", DataTypes.STRING())
                .column("url", DataTypes.STRING())
                .build();
        tEnv.createTemporaryTable("sinkTable", TableDescriptor.forConnector("print")
                .schema(schema)
                .build());
        //4、Table API和SQL混用
        Table resultTable = tEnv.from("sourceTable").select($("user"), $("url"));
        tEnv.createTemporaryView("resultTableView", resultTable);
        Table result = tEnv.sqlQuery("select * from resultTableView where user = 'Mary'");
        //5、输出(包括执行,不需要单独在调用tEnv.execute("job"))
        result.executeInsert("sinkTable");
        result.printSchema();
    }
}

Table API 和DataStream混用

package com.blink.sb;
import com.blink.sb.beans.ClickLogs;
import org.apache.flink.streaming.api.datastream.DataStream;
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 static org.apache.flink.table.api.Expressions.$;
public class MixFlinkTableAndDataStream {
    public static void main(String[] args) throws Exception {
        //1、获取Stream执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2、创建表执行环境
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        //3、读取数据(source)
        DataStream<ClickLogs> clickLogs = env.fromElements(
                "Mary,./home,2022-02-02 12:00:00",
                "Bob,./cart,2022-02-02 12:00:00",
                "Mary,./prod?id=1,2022-02-02 12:00:05",
                "Liz,./home,2022-02-02 12:01:00",
                "Bob,./prod?id=3,2022-02-02 12:01:30",
                "Mary,./prod?id=7,2022-02-02 12:01:45"
        ).map(event -> {
            String[] props = event.split(",");
            return ClickLogs
                    .builder()
                    .user(props[0])
                    .url(props[1])
                    .cTime(props[2])
                    .build();
        });
        //4、流转换为动态表
        Table table = tEnv.fromDataStream(clickLogs);
        //5、执行Table API查询/SQL查询
        Table resultTable = table
                .where($("user").isEqual("Mary"))
                .select($("user"), $("url"), $("cTime"));
        //6、将Table转换为DataStream
        DataStream<ClickLogs> selectedClickLogs = tEnv.toDataStream(resultTable, ClickLogs.class);
        //7、处理结果：打印/输出
        selectedClickLogs.print();
        //8、执行
        env.execute("FlinkTableFirstExample");
    }
}

Table api Connectors （内置连接器）

Flink 的 Table API & SQL通过Connectors连接外部系统，并执⾏批/流⽅式的读写操作。Connectors提供了丰富的外部系统连接器，根据source和sink的类型，它们⽀持不同的格式，例如 CSV、Avro、Parquet 或 ORC。⽬前⽀持的内置Connectors如下表所示（截⽌到Flink 1.14.3）：

注意：如果作为sink⼤家还要注意⽀持的输出模式（Append/Retract/Upsert）

参考:https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/table/overview/

Table API 输入(kafka)和输出(kafka)

package com.blink.sb;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.types.Row;
import java.util.concurrent.ExecutionException;
import static org.apache.flink.table.api.Expressions.$;
public class FlinkTableAPIKafka2Kafka {
    public static final String input_topic = "clicklog_input";
    public static final String output_topic = "clicklog_output";
    public static final String output_topic_change = "clicklog_output_change";
    public static void main(String[] args) throws ExecutionException, InterruptedException {
//1、创建TableEnvironment
        final ParameterTool parameterTool = ParameterTool.fromArgs(args);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                .build();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);
        //2、创建kafka source table
        final Schema schema = Schema.newBuilder()
                .column("user", DataTypes.STRING())
                .column("url", DataTypes.STRING())
                .column("cTime", DataTypes.STRING())
                .build();
        tEnv.createTemporaryTable("sourceTable", TableDescriptor.forConnector("kafka")
                .schema(schema)
                .format("json")
                .option("topic", input_topic)
                .option("properties.bootstrap.servers", "node02:6667")
                .option("properties.group.id", "testGroup")//每次都从最早的offsets开始
                .option("scan.startup.mode", "latest-offset")
                .build());
        //3、创建kafka sink table
        final Schema sinkSchema = Schema.newBuilder()
                .column("user", DataTypes.STRING())
                .column("url", DataTypes.STRING())
                .column("cTime", DataTypes.STRING())
//                .column("cnt", DataTypes.BIGINT())
                .build();
        tEnv.createTemporaryTable("sinkTable", TableDescriptor.forConnector("kafka")
                .schema(sinkSchema)
                .format("csv")
                .option("topic", output_topic)
                .option("properties.bootstrap.servers", "node02:6667")
                .build());
        //4、Table API从sourceTable读取数据并以csv格式写入sinkTable
        /**
         *
         * 如果报错：Could not find any factory for identifier 'kafka' that implements 'org.apache.flink.table.factories.DynamicTableFactory' in the classpath
         * 原因：缺省 flink-connector-kafka_${scala.binary.version}  这个库的支持。 这个库支持 upsert-kafka
         * 请查看 https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/connectors/table/upsert-kafka/
         *
         * 这种模式 insert_only
         */
        tEnv.from("sourceTable")
                .select($("user"), $("url"), $("cTime"))
                .executeInsert("sinkTable");
        /**
         * Table sink 'default_catalog.default_database.sinkTable' doesn't support consuming update changes which is produced by node GroupAggregate(groupBy=[user, url], select=[user, url, COUNT(url) AS EXPR$0])
         * 向sinkTable中注入数据时， 不支持 分组聚合的table
         *
         * 解决方式： 将聚合类型的table 转换为 ChangelogStream类型的<>DataStream</>， 然后，DataStream.sink 到 kafka 中
         */
        Table select = tEnv.from("sourceTable")
                .groupBy($("user"), $("url"))
                .select($("user"), $("url").count().as("cnt"));
        DataStream<Row> rowDataStream = tEnv.toChangelogStream(select,
                Schema.newBuilder()
                        .column("user", DataTypes.STRING())
                        .column("cnt", DataTypes.BIGINT()).build(),
                ChangelogMode.all());
        DataStream<String> stringDataStream = rowDataStream.map(new MapFunction<Row, String>() {
            @Override
            public String map(Row row) throws Exception {
                return "xxxx";
            }
        });
        stringDataStream.sinkTo(KafkaSink.<String>builder()
                .setBootstrapServers("node02:6667")
                .setRecordSerializer(KafkaRecordSerializationSchema.builder()
                        .setTopic(output_topic_change)
                        .setKafkaValueSerializer(StringSerializer.class)
                        .build())
                .build());
    }
}

kafka输入和输出高级特性(更多option)

可以通过更多option来启⽤⾼级特性，kafka connector⽀持的option

详⻅如下链接： https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/connectors/table/kafka/#connector-options

key和value格式

Kafka消息的key和value均可指定format。

仅指定value format

kafka消息中key是可以选的，可以只指定value format

tEnv.createTemporaryTable("sourceTable", TableDescriptor.forConnector("kafka")
 .schema(schema)
 .format("json")
 ...
 .build());

指定key和value format

tEnv.createTemporaryTable("sourceTable", TableDescriptor.forConnector("kafka")
 .schema(schema)
// .format("json")
 .option("topic",input_topic)
 .option("key.format","json")
 .option("value.format","json")
 ...
 .build())

注意：format("json")和option("value.format","json")⼆选⼀，⼆者等价

Topic和Partition发现

可以通过topic或者topic-pattern来配置主题。

配置项	source	sink
topic	⽀持单个topic或者topic列表，topic列表⽤分号分割：topic-1;topic-2，topic和topic-pattern⼆选⼀	注意：作为sink时不⽀持topic列表
topic-pattern	通过topic名字的正则表达式来指定多个topic，例如：test-topic-[0- 9]，topic和topic-pattern⼆选⼀	不能作⽤于sink

注意：要允许在作业开始运⾏后发现动态创建的topic，请为 scan.topic-partition-discovery.interval 设置⼀个⾮负值

source读取位置

作为source是，可以通过scan.startup.mode选项指定从哪个位置开始消费，可选的值如下

选项值	说明
group-offsets（默认值）	使⽤当前消费组提交到ZK / Kafka的offset开始消费，Flink1.14有bug:https://issues.apach e.org/jira/browse/FLINK-2469
earliest-offset	从earliest offset开始消费（每次）
latest-offset	从latest offset开始消费（每次）
specific-offsets	从⽤户指定的每个分区的特定偏移量开始消费，必须跟scan.startup.specific-offsets配置搭配使⽤：例如：partition:0,offset:42;partition:1,offset:300

sink分区

当kafka作为sink时，可以通过sink.partitioner指定partitioner。

选项值	说明
default	使用kafka默认的partitioner
fixed	每个Flink分区上的消息最多分区到一个kafka分区，可以降低网络开销
round-robin	只有当kafka消息没有key时，才有用
自定义FlinkKafkaPartitioner的子类

一致性保证

默认情况下，如果启⽤checkpoint，Kafka sink使⽤at-least-once⼀致性语意。在启⽤checkpoint的前题下，可通过sink.delivery-guarantee来调整⼀致性语意：

选项值	说明
none	Flink不做任何保证，生成的记录可能会丢失或重复
at-least-once(default)	至少一次保证，但可能会重复
exactly-once	使用kafka事务提供精确一次语义

⼀旦启⽤了事物来保证exactly-once语意，⼀定要注意下游消费者要配置isolation.level为read_committed（默认是read_uncommitted）。

flink-sql(table api 编程)

Table API 标准结构

EXP

两种TableEnvironment

1、TableEnvironment

2、StreamTableEnvironment

TableEnviromnet的配置

创建表和注册表

创建表的两种⽅式

创建虚拟表

创建常规表

临时表和持久表

EXP

表的标识符

EXP

Table API 查询

常规查询

混⽤Table API和SQL查询

Table API 和DataStream混用

Table api Connectors （内置连接器）

Table API 输入(kafka)和输出(kafka)

kafka输入和输出高级特性(更多option)

key和value格式

Topic和Partition发现

source读取位置

sink分区

一致性保证

热门文章

最新文章

相关课程

相关电子书

探索云世界

热门

云计算

大数据

云原生

人工智能

数据库

开发与运维

活动广场

任务中心

训练营

直播

乘风者计划

下载

镜像站

技术资料

flink-sql(table api 编程)

Table API 标准结构

EXP

两种TableEnvironment

1、TableEnvironment

2、StreamTableEnvironment

TableEnviromnet的配置

创建表和注册表

创建表的两种⽅式

创建虚拟表

创建常规表

临时表和持久表

EXP

表的标识符

EXP

Table API 查询

常规查询

混⽤Table API和SQL查询

Table API 和DataStream混用

Table api Connectors （内置连接器）

Table API 输入(kafka)和输出(kafka)

kafka输入和输出高级特性(更多option)

key和value格式

Topic和Partition发现

source读取位置

sink分区

一致性保证

热门文章

最新文章

相关课程

相关电子书