当前位置: 首页>编程语言>正文

flinksql 连接数据库超时是什么原因 flink1.11 sql


Flink(七)Flink SQL

  • 一.SQL基础
  • 1.基础架构
  • 2.表环境
  • 2.1 表环境的作用
  • 2.2 创建表环境
  • 3.创建表
  • 3.1 表的ID
  • 3.2 DDL
  • 2.3 流转换成表
  • 4.表的查询
  • 4.1. 执行 SQL 进行查询
  • 4.2 调用 Table API 进行查询
  • 5.输出表
  • 5.1 executeInsert()
  • 5.2 表转换成流
  • 5.3 动态表和持续查询
  • 二.时间窗口 SQL
  • 1.事件时间
  • 1.1 DDL中定义
  • 1.2 流定义时间
  • 2.处理时间
  • 3.窗口
  • 4.聚合查询
  • 4.1 分组聚合
  • 4.2 窗口聚合
  • 5.开窗聚合
  • 6.Top N
  • 6.1 普通Top N
  • 6.2 窗口Top N
  • 7.Join
  • 三.函数
  • 3.1 系统函数
  • 3.2 UDF
  • 四.SQL 客户端
  • 五.连接外部系统
  • 1.文件系统
  • 2.Kafka
  • 3.JDBC
  • 4.Elasticsearch
  • 5.Hbase
  • 6.Hive


一.SQL基础

我们想要在代码中使用 Table API,必须引入相关的依赖

<dependency>
	 <groupId>org.apache.flink</groupId>
	 <artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId>
	 <version>${flink.version}</version>
</dependency>

这里的依赖是一个 Java 的“桥接器”(bridge),主要就是负责 Table API 和下层 DataStream API 的连接支持,按照不同的语言分为 Java 版和 Scala 版

如果我们希望在本地的集成开发环境(IDE)里运行 Table API 和 SQL,还需要引入以下依赖:

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId>
  <version>${flink.version}</version>
</dependency>
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
  <version>${flink.version}</version>
</dependency>
1.基础架构

在 Flink 中,Table API 和 SQL 可以看作联结在一起的一套 API,这套 API 的核心概念就是“表”(Table)。在我们的程序中,输入数据可以定义成一张表;然后对这张表进行查询,就可以得到新的表,这相当于就是流数据的转换操作;最后还可以定义一张用于输出的表,负责将处理结果写入到外部系统

我们可以看到,程序的整体处理流程与 DataStream API 非常相似,也可以分为读取数据源(Source)、转换(Transform)、输出数据(Sink)三部分;只不过这里的输入输出操作不需要额外定义,只需要将用于输入和输出的表定义出来,然后进行转换查询就可以了

程序基本架构如下:

// 创建表环境
TableEnvironment tableEnv = ...;
// 创建输入表,连接外部系统读取数据
tableEnv.executeSql("CREATE TEMPORARY TABLE inputTable ... WITH ( 'connector' 
= ... )");
// 注册一个表,连接到外部系统,用于输出
tableEnv.executeSql("CREATE TEMPORARY TABLE outputTable ... WITH ( 'connector' 
= ... )");
// 执行 SQL 对表进行查询转换,得到一个新的表
Table table1 = tableEnv.sqlQuery("SELECT ... FROM inputTable... ");
// 使用 Table API 对表进行查询转换,得到一个新的表
Table table2 = tableEnv.from("inputTable").select(...);
// 将得到的结果写入输出表
TableResult tableResult = table1.executeInsert("outputTable");

通过执行 DDL 来直接创建一个表。这里执行的 CREATE 语句中用 WITH 指定了外部系统的连接器,于是就可以连接外部系统读取数据了。这其实是更加一般化的程序架构,因为这样我们就可以完全抛开DataStream API,直接用 SQL 语句实现全部的流处理过程

而后面对于输出表的定义是完全一样的。可以发现,在创建表的过程中,其实并不区分“输入”还是“输出”,只需要将这个表“注册”进来、连接到外部系统就可以了;这里的 inputTable、outputTable 只是注册的表名,并不代表处理逻辑,可以随意更换。至于表的具体作用,则要等到执行后面的查询转换操作时才能明确。我们直接从 inputTable 中查询数据,那么 inputTable就是输入表;而 outputTable 会接收另外表的结果进行写入,那么就是输出表

2.表环境
2.1 表环境的作用

对于 Flink 这样的流处理框架来说,数据流和表在结构上还是有所区别的。所以使用 Table API 和 SQL 需要一个特别的运行时环境,这就是所谓的“表环境”(TableEnvironment)。它主要负责:

  • 注册 Catalog 和表
  • 执行 SQL 查询
  • 注册用户自定义函数(UDF)
  • DataStream 和表之间的转换
2.2 创建表环境

方法一 基于Stream流的环境(推荐)

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

方法二 脱离流式环境

EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()  // 可选择流处理或者批处理
                .useBlinkPlanner() // 计划器默认使用 blink planner
                .build();

TableEnvironment environment = TableEnvironment.create(settings);
3.创建表
3.1 表的ID

表都是通过 Catalog 来进行注册创建的。表在环境中有一个唯一的 ID,由三部分组成:目录(catalog)名,数据库(database)名,以及表名。在默认情况下,目录名为 default_catalog,数据库名为default_database。所以如果我们直接创建一个叫作 MyTable 的表,它的 ID 就是:

default_catalog.default_database.MyTable

如果希望使用自定义的目录名和库名,可以在环境中进行设置,前提是要在表环境中进行了注册

tEnv.useCatalog("custom_catalog");
tEnv.useDatabase("custom_database");
3.2 DDL

通过连接器(connector)连接到一个外部系统,然后定义出对应的表结构。例如我们可以连接到 Kafka 或者文件系统,将存储在这些外部系统的数据以“表”的形式定义出来,这样对表的读写就可以通过连接器转换成对外部系统的读写了。当我们在表环境中读取这张表,连接器就会从外部系统读取数据并进行转换;而当我们向这张表写入数据,连接器就会将数据输出(Sink)到外部系统中

在代码中,我们可以调用表环境的 executeSql()方法,可以传入一个 DDL 作为参数执行SQL 操作。这里我们传入一个 CREATE 语句进行表的创建,并通过 WITH 关键字指定连接到外部系统的连接器:

tableEnv.executeSql("CREATE [TEMPORARY] TABLE MyTable ... WITH ( 'connector' 
= ... )");

以下表DDL指明了连接器是一个文件,指明了文件路径,文件的格式

String createDDL = "CREATE TABLE clickTable (user_name string,url string,ts bigint) " +
                "with ('connector' = 'filesystem','path' = 'input/clicks.txt','format' ='csv')";
        
tableEnv.executeSql(createDDL);

// 我们可以调用from API 转换成Table对象,从而调用Table API
Table clickTable = tableEnv.from("clickTable");
2.3 流转换成表

我们之前的所有学习,都是基于流,如果我们能将流转换成表,那太方便了!!!

(1)调用 fromDataStream()方法

想要将一个 DataStream 转换成表也很简单,可以通过调用表环境的 fromDataStream()方法来实现,返回的就是一个 Table 对象

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 获取表环境
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
// 读取数据源
SingleOutputStreamOperator<Event> eventStream = env.addSource(...)
// 将数据流转换成表
Table eventTable = tableEnv.fromDataStream(eventStream);

如果流里是POJO对象,那么表的一行就是一个对象,表的列名就是对象的属性名,当然,我们可以自己选择对象的部分属性来组成表,当然,我们也可以使用as进行重命名

Table table = tableEnv.fromDataStream(stream, $("user").as("myUser"), $("url").as("myUrl"));

但是,我们这样得到的是一个Table对象!想要进行SQL操作还需要使用以下方法转换成临时视图

flinksql 连接数据库超时是什么原因 flink1.11 sql,flinksql 连接数据库超时是什么原因 flink1.11 sql_字段,第1张

(2)调用 createTemporaryView()方法

调用 fromDataStream()方法简单直观,可以直接实现 DataStream 到 Table 的转换;不过如果我们希望直接在 SQL 中引用这张表,就还需要调用表环境的 createTemporaryView()方法来创建虚拟视图了

对于这种场景,也有一种更简洁的调用方式。我们可以直接调用 createTemporaryView()方法创建虚拟表,传入的两个参数,第一个依然是注册的表名,而第二个可以直接就是DataStream。之后仍旧可以传入多个参数,用来指定表中的字段

tableEnv.createTemporaryView("EventTable", eventStream, $("timestamp").as("ts"),$("url"));

flinksql 连接数据库超时是什么原因 flink1.11 sql,flinksql 连接数据库超时是什么原因 flink1.11 sql_SQL_02,第2张

(3)调用 fromChangelogStream ()方法

表环境还提供了一个方法 fromChangelogStream(),可以将一个更新日志流转换成表。这个方法要求流中的数据类型只能是 Row,而且每一个数据都需要指定当前行的更新类型(RowKind);所以一般是由连接器帮我们实现的,直接应用比较少见

流转换成表支持的数据类型

前面示例中的 DataStream,流中的数据类型都是定义好的 POJO 类。如果 DataStream 中的类型是简单的基本类型,还可以直接转换成表吗?这就涉及了 Table 中支持的数据类型。整体来看,DataStream 中支持的数据类型,Table 中也是都支持的,只不过在进行转换时需要注意一些细节

(1)原子类型

在 Flink 中,基础数据类型(Integer、Double、String)和通用数据类型(也就是不可再拆分的数据类型)统一称作“原子类型”。原子类型的 DataStream,转换之后就成了只有一列的Table,列字段(field)的数据类型可以由原子类型推断出。另外,还可以在 fromDataStream()方法里增加参数,用来重新命名列字段

StreamTableEnvironment tableEnv = ...;
DataStream<Long> stream = ...;
// 将数据流转换成动态表,动态表只有一个字段,重命名为 myLong
Table table = tableEnv.fromDataStream(stream, $("myLong"));

(2)Tuple 类型

当原子类型不做重命名时,默认的字段名就是“f0”,容易想到,这其实就是将原子类型看作了一元组 Tuple1 的处理结果

Table 支持 Flink 中定义的元组类型 Tuple,对应在表中字段名默认就是元组中元素的属性名 f0、f1、f2…。所有字段都可以被重新排序,也可以提取其中的一部分字段。字段还可以通过调用表达式的 as()方法来进行重命名

StreamTableEnvironment tableEnv = ...;
DataStream<Tuple2<Long, Integer>> stream = ...;
// 将数据流转换成只包含 f1 字段的表
Table table = tableEnv.fromDataStream(stream, $("f1"));
// 将数据流转换成包含 f0 和 f1 字段的表,在表中 f0 和 f1 位置交换
Table table = tableEnv.fromDataStream(stream, $("f1"), $("f0"));
// 将 f1 字段命名为 myInt,f0 命名为 myLong
Table table = tableEnv.fromDataStream(stream, $("f1").as("myInt"), $("f0").as("myLong"));

(3)POJO 类型

Flink 也支持多种数据类型组合成的“复合类型”,最典型的就是简单 Java 对象(POJO 类型)。由于 POJO 中已经定义好了可读性强的字段名,这种类型的数据流转换成 Table 就显得无比顺畅了

将 POJO 类型的 DataStream 转换成 Table,如果不指定字段名称,就会直接使用原始 POJO 类型中的字段名称。POJO 中的字段同样可以被重新排序、提却和重命名

StreamTableEnvironment tableEnv = ...;
DataStream<Event> stream = ...;
Table table = tableEnv.fromDataStream(stream);
Table table = tableEnv.fromDataStream(stream, $("user"));
Table table = tableEnv.fromDataStream(stream, $("user").as("myUser"), $("url").as("myUrl"));

(4)Row 类型

Flink 中还定义了一个在关系型表中更加通用的数据类型——行(Row),它是 Table 中数据的基本组织形式。Row 类型也是一种复合类型,它的长度固定,而且无法直接推断出每个字段的类型,所以在使用时必须指明具体的类型信息;我们在创建 Table 时调用的 CREATE语句就会将所有的字段名称和类型指定,这在 Flink 中被称为表的“模式结构”(Schema)。除此之外,Row 类型还附加了一个属性 RowKind,用来表示当前行在更新操作中的类型。这样,Row 就可以用来表示更新日志流(changelog stream)中的数据,从而架起了 Flink 中流和表的
转换桥梁

所以在更新日志流中,元素的类型必须是 Row,而且需要调用 ofKind()方法来指定更新类型

DataStream<Row> dataStream =
 env.fromElements(
 Row.ofKind(RowKind.INSERT, "Alice", 12),
 Row.ofKind(RowKind.INSERT, "Bob", 5),
 Row.ofKind(RowKind.UPDATE_BEFORE, "Alice", 12),
 Row.ofKind(RowKind.UPDATE_AFTER, "Alice", 100));
// 将更新日志流转换为表
Table table = tableEnv.fromChangelogStream(dataStream);
4.表的查询

Flink 为我们提供了两种查询方式:SQL 和 Table API

4.1. 执行 SQL 进行查询

在代码中,我们只要调用表环境的 sqlQuery()方法,传入一个字符串形式的 SQL 查询语句就可以了。执行得到的结果,是一个 Table 对象

// 创建表环境
TableEnvironment tableEnv = ...; 
// 创建表
tableEnv.executeSql("CREATE TABLE EventTable ... WITH ( 'connector' = ... )");
// 查询用户 Alice 的点击事件,并提取表中前两个字段
Table aliceVisitTable = tableEnv.sqlQuery(
 "SELECT user, url " +
 "FROM EventTable " +
 "WHERE user = 'Alice' "
 );

我们也可以通过 GROUP BY 关键字定义分组聚合,调用 COUNT()、SUM()这样的函数来进行统计计算:

Table urlCountTable = tableEnv.sqlQuery(
 "SELECT user, COUNT(url) " +
 "FROM EventTable " +
 "GROUP BY user "
 );

上面的例子得到的是一个新的 Table 对象,我们可以再次将它注册为虚拟表继续在 SQL中调用。另外,我们也可以直接将查询的结果写入到已经注册的表中,这需要调用表环境的executeSql()方法来执行 DDL,传入的是一个 INSERT 语句:

// 注册表
tableEnv.executeSql("CREATE TABLE EventTable ... WITH ( 'connector' = ... )");
tableEnv.executeSql("CREATE TABLE OutputTable ... WITH ( 'connector' = ... )");
// 将查询结果输出到 OutputTable 中
tableEnv.executeSql (
"INSERT INTO OutputTable " +
 "SELECT user, url " +
 "FROM EventTable " +
 "WHERE user = 'Alice' "
 );
4.2 调用 Table API 进行查询

另外一种查询方式就是调用 Table API。这是嵌入在 Java 和 Scala 语言内的查询 API,核心就是 Table 接口类,通过一步步链式调用 Table 的方法,就可以定义出所有的查询转换操作

每一步方法调用的返回结果,都是一个 Table

由于Table API是基于Table的Java实例进行调用的,因此我们首先要得到表的Java对象。基于环境中已注册的表,可以通过表环境的 from()方法非常容易地得到一个 Table 对象:

Table eventTable = tableEnv.from("EventTable");

传入的参数就是注册好的表名。注意这里 eventTable 是一个 Table 对象,而 EventTable 是在环境中注册的表名。得到 Table 对象之后,就可以调用 API 进行各种转换操作了,得到的是一个新的 Table 对象:

Table maryClickTable = eventTable
 .where($("user").isEqual("Alice"))
 .select($("url"), $("user"));

这里每个方法的参数都是一个“表达式”(Expression),用方法调用的形式直观地说明了想要表达的内容;“$”符号用来指定表中的一个字段。上面的代码和直接执行 SQL 是等效的。Table API 是嵌入编程语言中的 DSL,SQL 中的很多特性和功能必须要有对应的实现才可以使用,因此跟直接写 SQL 比起来肯定就要麻烦一些。目前 Table API 支持的功能相对更少,可以预见未来 Flink 社区也会以扩展 SQL 为主,为大家提供更加通用的接口方式;所以我们接下来也会以介绍 SQL 为主,简略地提及 Table API

5.输出表

表的创建和查询,就对应着流处理中的读取数据源(Source)和转换(Transform);而最后一个步骤 Sink,也就是将结果数据输出到外部系统,就对应着表的输出操作

5.1 executeInsert()

在代码上,输出一张表最直接的方法,就是调用 Table 的方法 executeInsert()方法将一个Table 写入到注册过的表中,方法传入的参数就是注册的表名

// 注册表,用于输出数据到外部系统
tableEnv.executeSql("CREATE TABLE OutputTable ... WITH ( 'connector' = ... )");
// 经过查询转换,得到结果表
Table result = ...
// 将结果表写入已注册的输出表中
result.executeInsert("OutputTable");

在底层,表的输出是通过将数据写入到 TableSink 来实现的。TableSink 是 Table API 中提供的一个向外部系统写入数据的通用接口,可以支持不同的文件格式(比如 CSV、Parquet)、存储数据库(比如 JDBC、HBase、Elasticsearch)和消息队列(比如 Kafka)。它有些类似于DataStream API 中调用 addSink()方法时传入的 SinkFunction,有不同的连接器对它进行了实现

5.2 表转换成流

(1)调用 toDataStream()方法

将一个 Table 对象转换成 DataStream 非常简单,只要直接调用表环境的方法 toDataStream()就可以了。得到的流只是一个仅追加流(只有插入),只有插入操作

// 这里需要将要转换的 Table 对象作为参数传入
tableEnv.toDataStream(myTable).print();

(2)调用 toChangelogStream()方法

对于有更新操作的表,我们不要试图直接把它转换成 DataStream 打印输出,而是记录一下它的“更新日志”(change log)。这样一来,对于表的所有更新操作,就变成了一条更新日志的流,我们就可以转换成流打印输出了。具体的编码规则是:INSERT 插入操作编码为 add 消息;DELETE 删除操作编码为 retract消息;而 UPDATE 更新操作则编码为被更改行的 retract 消息,和更新后行(新行)的 add 消息。这样,我们可以通过编码后的消息指明所有的增删改操作,一个动态表就可以转换为撤回流(add + retract)

tableEnv.toChangelogStream(urlCountTable).print();

更新插入流(upsert)中只包含两种类型的消息:更新插入(upsert)消息和删除(delete)消息。所谓的“upsert”其实是“update”和“insert”的合成词,所以对于更新插入流来说,INSERT 插入操作和UPDATE更新操作,统一被编码为upsert消息;而DELETE删除操作则被编码为delete消息

既然更新插入流中不区分插入(insert)和更新(update),那我们自然会想到一个问题:如果希望更新一行数据时,怎么保证最后做的操作不是插入呢?
这就需要动态表中必须有唯一的键(key)。通过这个 key 进行查询,如果存在对应的数据就做更新(update),如果不存在就直接插入(insert)。这是一个动态表可以转换为更新插入流的必要条件。当然,收到这条流中数据的外部系统,也需要知道这唯一的键(key),这样才能正确地处理消息

需要注意的是,在代码里将动态表转换为 DataStream 时,只支持仅追加(append-only)和撤回(retract)流,我们调用 toChangelogStream()得到的其实就是撤回流;这也很好理解,DataStream 中并没有 key 的定义,所以只能通过两条消息一减一增来表示更新操作。而连接到外部系统时,则可以支持不同的编码方法,这取决于外部系统本身的特性

5.3 动态表和持续查询

当流中有新数据到来,初始的表中会插入一行;而基于这个表定义的 SQL 查询,就应该在之前的基础上更新结果。这样得到的表就会不断地动态变化,被称为“动态表”(Dynamic Tables)

动态表可以像静态的批处理表一样进行查询操作。由于数据在不断变化,因此基于它定义的 SQL 查询也不可能执行一次就得到最终结果。这样一来,我们对动态表的查询也就永远不会停止,一直在随着新数据的到来而继续执行。这样的查询就被称作“持续查询”(Continuous Query)。对动态表定义的查询操作,都是持续查询;而持续查询的结果也会是一个动态表

如下图,就是持续查询更新动态表的案例

flinksql 连接数据库超时是什么原因 flink1.11 sql,flinksql 连接数据库超时是什么原因 flink1.11 sql_SQL_03,第3张

我们可以看到,结果前面出现了 +I +U -U,以+I 为前缀,表示都是以 INSERT 操作追加到结果表中的,-U表示更新前的状态,+U表示更新后的状态,于是,当一种查询只包含+I操作是,我们叫做追加查询。当一种查询包含I、U时,称为更新查询

flinksql 连接数据库超时是什么原因 flink1.11 sql,flinksql 连接数据库超时是什么原因 flink1.11 sql_flink_04,第4张

二.时间窗口 SQL

1.事件时间
1.1 DDL中定义

在创建表的 DDL(CREATE TABLE 语句)中,可以增加一个字段,通过 WATERMARK语句来定义事件时间属性。WATERMARK 语句主要用来定义水位线(watermark)的生成表达式,这个表达式会将带有事件时间戳的字段标记为事件时间属性,并在它基础上给出水位线的延迟时间。具体定义方式如下:

CREATE TABLE EventTable(
 user STRING,
 url STRING,
 ts TIMESTAMP(3), // 单位是毫秒
 WATERMARK FOR ts AS ts - INTERVAL '5' SECOND // 水位线延迟5秒
) WITH (
 ...
);

这里我们把 ts 字段定义为事件时间属性,而且基于 ts 设置了 5 秒的水位线延迟。这里的“5 秒”是以“时间间隔”的形式定义的,格式是 INTERVAL <数值> <时间单位>:INTERVAL '5' SECOND

这里的数值必须用单引号引起来,而单位用 SECOND 和 SECONDS 是等效的。Flink 中支持的事件时间属性数据类型必须为 TIMESTAMP 或者 TIMESTAMP_LTZ。这里TIMESTAMP_LTZ 是指带有本地时区信息的时间戳(TIMESTAMP WITH LOCAL TIME ZONE);一般情况下如果数据中的时间戳是“年-月-日-时-分-秒”的形式,那就是不带时区信息的,可以将事件时间属性定义为 TIMESTAMP 类型

而如果原始的时间戳就是一个长整型的毫秒数,这时就需要另外定义一个字段来表示事件时间属性,类型定义为 TIMESTAMP_LTZ 会更方便:

CREATE TABLE events (
 user STRING,
 url STRING,
 ts BIGINT,
 ts_ltz AS TO_TIMESTAMP_LTZ(ts, 3),
 WATERMARK FOR ts_ltz AS ts_ltz - INTERVAL '5' SECOND
) WITH (
 ...
);
1.2 流定义时间

事件时间属性也可以在将 DataStream 转换为表的时候来定义。我们调用 fromDataStream()方法创建表时,可以追加参数来定义表中的字段结构;这时可以给某个字段加上.rowtime() 后缀,就表示将当前字段指定为事件时间属性。这个字段可以是数据中本不存在、额外追加上去的“逻辑字段”,就像之前 DDL 中定义的第二种情况;也可以是本身固有的字段,那么这个字段就会被事件时间属性所覆盖,类型也会被转换为 TIMESTAMP。不论那种方式,时间属性字段中保存的都是事件的时间戳(TIMESTAMP 类型)

需要注意的是,这种方式只负责指定时间属性,而时间戳的提取和水位线的生成应该之前就在 DataStream 上定义好了。由于 DataStream 中没有时区概念,因此 Flink 会将事件时间属性解析成不带时区的 TIMESTAMP 类型,所有的时间值都被当作 UTC 标准时间

// 1.创建一个表执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

env.setParallelism(1);
env.getConfig().setAutoWatermarkInterval(100); // 100毫秒生成一次水位线

SingleOutputStreamOperator<Event> streamOperator = env.addSource(new ClickSource())
        // 乱序流的WaterMark生成
        .assignTimestampsAndWatermarks(WatermarkStrategy
                .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2)) // 延迟2秒保证数据正确
                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                    @Override // 时间戳的提取器
                    public long extractTimestamp(Event event, long l) {
                        return event.getTimestamp();
                    }
                })
        );

tableEnv.fromDataStream(streamOperator,$("user_name"),$("url"),$("timestamp").as ("ts")
,$("et").rowtime());

flinksql 连接数据库超时是什么原因 flink1.11 sql,flinksql 连接数据库超时是什么原因 flink1.11 sql_flink_05,第5张

因此这个表的4个列是 用户名、url、用户时间戳、事件时间

注意:调用了rowtime方法就会把长整型的时间戳转换成timestamp,因此用户时间戳解析成timestamp就是事件时间,因此如果我们不想要用户时间戳这个信息,直接让用户时间戳调用rowtime方法,就让表中有了事件时间了,此时表中是3列了

2.处理时间

相比之下处理时间就比较简单了,它就是我们的系统时间,使用时不需要提取时间戳(timestamp)和生成水位线(watermark)。因此在定义处理时间属性时,必须要额外声明一个字段,专门用来保存当前的处理时间

1. 在创建表的 DDL 中定义

在创建表的 DDL(CREATE TABLE 语句)中,可以增加一个额外的字段,通过调用系统内置的 PROCTIME()函数来指定当前的处理时间属性,返回的类型是 TIMESTAMP_LTZ

CREATE TABLE EventTable(
 user STRING,
 url STRING,
 ts AS PROCTIME()
) WITH (
 ...
);

这里的时间属性,其实是以“计算列”(computed column)的形式定义出来的。所谓的计算列是 Flink SQL 中引入的特殊概念,可以用一个 AS 语句来在表中产生数据中不存在的列,并且可以利用原有的列、各种运算符及内置函数。在前面事件时间属性的定义中,将 ts 字段转换成 TIMESTAMP_LTZ 类型的 ts_ltz,也是计算列的定义方式

2. 在数据流转换为表时定义

处 理 时 间 属 性 同 样 可 以 在 将 DataStream 转 换 为 表 的 时 候 来 定 义 。 我 们 调 用fromDataStream()方法创建表时,可以用.proctime()后缀来指定处理时间属性字段。由于处理时间是系统时间,原始数据中并没有这个字段,所以处理时间属性一定不能定义在一个已有字段上,只能定义在表结构所有字段的最后,作为额外的逻辑字段出现

代码中定义处理时间属性的方法如下:

DataStream<Tuple2<String, String>> stream = ...;
// 声明一个额外的字段作为处理时间属性字段
Table table = tEnv.fromDataStream(stream, $("user"), $("url"), 
$("ts").proctime());
3.窗口

目前 Flink 提供了以下几个窗口 TVF:

  • 滚动窗口(Tumbling Windows)
  • 滑动窗口(Hop Windows,跳跃窗口)
  • 累积窗口(Cumulate Windows)
  • 会话窗口(Session Windows,目前尚未完全支持)

在窗口 TVF 的返回值中,除去原始表中的所有列,还增加了用来描述窗口的额外 3 个列:“窗口起始点”(window_start)、“窗口结束点”(window_end)、“窗口时间”(window_time)起始点和结束点比较好理解,这里的“窗口时间”指的是窗口中的时间属性,它的值等于window_end - 1ms,所以相当于是窗口中能够包含数据的最大时间戳

(1)滚动窗口(TUMBLE)

滚动窗口在 SQL 中的概念与 DataStream API 中的定义完全一样,是长度固定、时间对齐、无重叠的窗口,一般用于周期性的统计计算

在 SQL 中通过调用 TUMBLE()函数就可以声明一个滚动窗口,只有一个核心参数就是窗口大小(size)。在 SQL 中不考虑计数窗口,所以滚动窗口就是滚动时间窗口,参数中还需要将当前的时间属性字段传入;另外,窗口 TVF 本质上是表函数,可以对表进行扩展,所以还应该把当前查询的表作为参数整体传入。具体声明如下:

TUMBLE(TABLE EventTable, DESCRIPTOR(ts), INTERVAL '1' HOUR)

这里基于时间字段 ts,对表 EventTable 中的数据开了大小为 1 小时的滚动窗口。窗口会将表中的每一行数据,按照它们 ts 的值分配到一个指定的窗口中

(2)滑动窗口(HOP)

滑动窗口的使用与滚动窗口类似,可以通过设置滑动步长来控制统计输出的频率。在 SQL中通过调用 HOP()来声明滑动窗口;除了也要传入表名、时间属性外,还需要传入窗口大小(size)和滑动步长(slide)两个参数

HOP(TABLE EventTable, DESCRIPTOR(ts), INTERVAL '5' MINUTES, INTERVAL '1' HOURS));

这里我们基于时间属性 ts,在表 EventTable 上创建了大小为 1 小时的滑动窗口,每 5 分钟滑动一次。需要注意的是,紧跟在时间属性字段后面的第三个参数是步长(slide),第四个参数才是窗口大小(size)

(3)累积窗口(CUMULATE)

滚动窗口和滑动窗口,可以用来计算大多数周期性的统计指标。不过在实际应用中还会遇到这样一类需求:我们的统计周期可能较长,因此希望中间每隔一段时间就输出一次当前的统计值;与滑动窗口不同的是,在一个统计周期内,我们会多次输出统计值,它们应该是不断叠加累积的

例如,我们按天来统计网站的 PV(Page View,页面浏览量),如果用 1 天的滚动窗口,那需要到每天 24 点才会计算一次,输出频率太低;如果用滑动窗口,计算频率可以更高,但统计的就变成了“过去 24 小时的 PV”。所以我们真正希望的是,还是按照自然日统计每天的PV,不过需要每隔 1 小时就输出一次当天到目前为止的 PV 值。这种特殊的窗口就叫作“累积窗口”(Cumulate Window)

flinksql 连接数据库超时是什么原因 flink1.11 sql,flinksql 连接数据库超时是什么原因 flink1.11 sql_大数据_06,第6张

累积窗口是窗口 TVF 中新增的窗口功能,它会在一定的统计周期内进行累积计算。累积窗口中有两个核心的参数:最大窗口长度(max window size)和累积步长(step)。所谓的最大窗口长度其实就是我们所说的“统计周期”,最终目的就是统计这段时间内的数据

开始时,创建的第一个窗口大小就是步长 step;之后的每个窗口都会在之前的基础上再扩展 step 的长度,直到达到最大窗口长度。在 SQL 中可以用 CUMULATE()函数来定义,具体如下:

CUMULATE(TABLE EventTable, DESCRIPTOR(ts), INTERVAL '1' HOURS, INTERVAL '1' DAYS))

这里我们基于时间属性 ts,在表 EventTable 上定义了一个统计周期为 1 天、累积步长为 1小时的累积窗口。注意第三个参数为步长 step,第四个参数则是最大窗口长度

4.聚合查询
4.1 分组聚合

SQL 中一般所说的聚合我们都很熟悉,主要是通过内置的一些聚合函数来实现的,比如SUM()、MAX()、MIN()、AVG()以及 COUNT()。它们的特点是对多条输入数据进行计算,得到一个唯一的值,属于“多对一”的转换。比如我们可以通过下面的代码计算输入数据的个数:

Table eventCountTable = tableEnv.sqlQuery("select COUNT(*) from EventTable");

而更多的情况下,我们可以通过 GROUP BY 子句来指定分组的键(key),从而对数据按照某个字段做一个分组统计。例如之前我们举的例子,可以按照用户名进行分组,统计每个用户点击 url 的次数:

public class TimeAndWindow {
    public static void main(String[] args) throws Exception {
        // 1.创建一个表执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        env.setParallelism(1);
        env.getConfig().setAutoWatermarkInterval(100); // 100毫秒生成一次水位线

        SingleOutputStreamOperator<Event> streamOperator = env.addSource(new ClickSource())
                // 乱序流的WaterMark生成
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2)) // 延迟2秒保证数据正确
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @Override // 时间戳的提取器
                            public long extractTimestamp(Event event, long l) {
                                return event.getTimestamp();
                            }
                        })
                );

        Table clickTable = tableEnv.fromDataStream(streamOperator, $("user"), $("url"), $("timestamp").as("ts")
                , $("et").rowtime());
        // 将表注册到表环境中
        tableEnv.createTemporaryView("clickTable",clickTable);

        Table aggTable = tableEnv.sqlQuery("select user,count(url) from clickTable group by user");


        tableEnv.toChangelogStream(aggTable).print();

        env.execute();
    }
}

另外,在持续查询的过程中,由于用于分组的 key 可能会不断增加,因此计算结果所需要维护的状态也会持续增长。为了防止状态无限增长耗尽资源,Flink Table API 和 SQL 可以在表环境中配置状态的生存时间(TTL):

TableEnvironment tableEnv = ...
// 获取表环境的配置
TableConfig tableConfig = tableEnv.getConfig();
// 配置状态保持时间
tableConfig.setIdleStateRetention(Duration.ofMinutes(60));
或者也可以直接设置配置项 table.exec.state.ttl:

需要注意,配置 TTL 有可能会导致统计结果不准确,这其实是以牺牲正确性为代价换取了资源的释放

此外,在 Flink SQL 的分组聚合中同样可以使用 DISTINCT 进行去重的聚合处理;可以使用 HAVING 对聚合结果进行条件筛选;还可以使用 GROUPING SETS(分组集)设置多个分组情况分别统计。这些语法跟标准 SQL 中的用法一致,这里就不再详细展开了

4.2 窗口聚合
public class WindowAggTest {
    public static void main(String[] args) throws Exception {
        // 创建一个表执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        env.setParallelism(1);
        env.getConfig().setAutoWatermarkInterval(100); // 100毫秒生成一次水位线

        SingleOutputStreamOperator<Event> streamOperator = env.addSource(new ClickSource())
                // 乱序流的WaterMark生成
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2)) // 延迟2秒保证数据正确
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @Override // 时间戳的提取器
                            public long extractTimestamp(Event event, long l) {
                                return event.getTimestamp();
                            }
                        })
                );

        Table clickTable = tableEnv.fromDataStream(streamOperator, $("user"), $("url"), $("timestamp").as("ts")
                , $("et").rowtime()); // et是事件时间
        // 将表注册到表环境中
        tableEnv.createTemporaryView("clickTable",clickTable);
        // 1.滚动窗口
        Table tumbleWindowAgg = tableEnv.sqlQuery(
                "SELECT user,window_end AS endT, COUNT(url) AS cnt " +
                "FROM TABLE( TUMBLE( TABLE clickTable, DESCRIPTOR(et), INTERVAL '10' SECOND)) " +
                "GROUP BY user,window_start,window_end"
        );
        tableEnv.toChangelogStream(tumbleWindowAgg).print("tumble => ");

        // 2.滑动窗口
        Table hopWindowAgg = tableEnv.sqlQuery("select user, count(url) as cnt, window_end as endT " +
                "from TABLE(HOP(TABLE clickTable,DESCRIPTOR(et),INTERVAL '5' SECOND,INTERVAL '10' SECOND))" +
                "GROUP BY user,window_start,window_end"
        );
        tableEnv.toChangelogStream(hopWindowAgg).print("hop => ");

        // 3.累积窗口
        Table cumulateWindowAgg = tableEnv.sqlQuery("select user, count(url) as cnt, window_end as endT " +
                "from TABLE(CUMULATE(TABLE clickTable,DESCRIPTOR(et),INTERVAL '5' SECOND,INTERVAL '10' SECOND))" +
                "GROUP BY user,window_start,window_end"
        );
        tableEnv.toChangelogStream(cumulateWindowAgg).print("cumulateWindowAgg => ");



        env.execute();
    }
}
5.开窗聚合

在标准 SQL 中还有另外一类比较特殊的聚合方式,可以针对每一行计算一个聚合值。比如说,我们可以以每一行数据为基准,计算它之前 1 小时内所有数据的平均值;也可以计算它之前 10 个数的平均值。就好像是在每一行上打开了一扇窗户、收集数据进行统计一样,这就是所谓的“开窗函数”。开窗函数的聚合与之前两种聚合有本质的不同:分组聚合、窗口 TVF聚合都是“多对一”的关系,将数据分组之后每组只会得到一个聚合结果;而开窗函数是对每行都要做一次开窗聚合,因此聚合之后表中的行数不会有任何减少,是一个“多对多”的关系

与标准 SQL 中一致,Flink SQL 中的开窗函数也是通过 OVER 子句来实现的,所以有时开窗聚合也叫作“OVER 聚合”(Over Aggregation)。基本语法如下:

SELECT
 <聚合函数> OVER (
 [PARTITION BY <字段 1>[, <字段 2>, ...]]
 ORDER BY <时间属性字段>
 <开窗范围>),
 ...
FROM ...

这里 OVER 关键字前面是一个聚合函数,它会应用在后面 OVER 定义的窗口上。在 OVER子句中主要有以下几个部分:
PARTITION BY(可选)

用来指定分区的键(key),类似于 GROUP BY 的分组,这部分是可选的

ORDER BY
OVER 窗口是基于当前行扩展出的一段数据范围,选择的标准可以基于时间也可以基于数量。不论那种定义,数据都应该是以某种顺序排列好的;而表中的数据本身是无序的。所以在OVER 子句中必须用 ORDER BY 明确地指出数据基于那个字段排序。在 Flink 的流处理中,目前只支持按照时间属性的升序排列,所以这里 ORDER BY 后面的字段必须是定义好的时间属性

开窗范围

对于开窗函数而言,还有一个必须要指定的就是开窗的范围,也就是到底要扩展多少行来做聚合。这个范围是由 BETWEEN <下界> AND <上界> 来定义的,也就是“从下界到上界”的范围。目前支持的上界只能是 CURRENT ROW,也就是定义一个“从之前某一行到当前行”的范围,所以一般的形式为:

BETWEEN ... PRECEDING AND CURRENT ROW

前面我们提到,开窗选择的范围可以基于时间,也可以基于数据的数量。所以开窗范围还应该在两种模式之间做出选择:范围间隔(RANGE intervals)和行间隔(ROW intervals)

  • 范围间隔

范围间隔以 RANGE 为前缀,就是基于 ORDER BY 指定的时间字段去选取一个范围,一般就是当前行时间戳之前的一段时间。例如开窗范围选择当前行之前 1 小时的数据:

RANGE BETWEEN INTERVAL '1' HOUR PRECEDING AND CURRENT ROW
  • 行间隔
    行间隔以 ROWS 为前缀,就是直接确定要选多少行,由当前行出发向前选取就可以了。例如开窗范围选择当前行之前的 5 行数据(最终聚合会包括当前行,所以一共 6 条数据):
ROWS BETWEEN 5 PRECEDING AND CURRENT ROW
// 开窗函数
Table overWindow = tableEnv.sqlQuery("select user, avg(ts) over(" +
        " PARTITION BY user order by et rows between 3 preceding and current row) as avg_ts" +
        " from clickTable");

tableEnv.toChangelogStream(overWindow).print();

多个聚合函数共用一个窗口

开窗聚合与窗口聚合(窗口 TVF 聚合)本质上不同,不过也还是有一些相似之处的:它们都是在无界的数据流上划定了一个范围,截取出有限数据集进行聚合统计;这其实都是“窗 口”的思路。事实上,在 Table API 中确实就定义了两类窗口:分组窗口(GroupWindow)和开窗窗口(OverWindow);而在 SQL 中,也可以用 WINDOW 子句来在 SELECT 外部单独定义一个 OVER 窗口:

SELECT user, ts,
COUNT(url) OVER w AS cnt, MAX(CHAR_LENGTH(url)) OVER w AS max_url
FROM EventTable
WINDOW w AS (
PARTITION BY user
ORDER BY ts
ROWS BETWEEN 2 PRECEDING AND CURRENT ROW)
6.Top N
6.1 普通Top N

在 Flink SQL 中,是通过 OVER 聚合和一个条件筛选来实现 Top N 的。具体来说,是通过将一个特殊的聚合函数ROW_NUMBER()应用到OVER窗口上,统计出每一行排序后的行号,作为一个字段提取出来;然后再用 WHERE 子句筛选行号小于等于 N 的那些行返回
基本语法如下:

SELECT ...
FROM (
 SELECT ...,
 ROW_NUMBER() OVER (
[PARTITION BY <字段 1>[, <字段 1>...]]
 ORDER BY <排序字段 1> [asc|desc][, <排序字段 2> [asc|desc]...]
) AS row_num
 FROM ...)
WHERE row_num <= N [AND <其它条件>]

这里的 OVER 窗口定义与之前的介绍基本一致,目的就是利用 ROW_NUMBER()函数为每一行数据聚合得到一个排序之后的行号。行号重命名为 row_num,并在外层的查询中以row_num <= N 作为条件进行筛选,就可以得到根据排序字段统计的 Top N 结果了

之前介绍的 OVER 窗口不是说了,目前 ORDER BY 后面只能跟时间字段、并且只支持升序吗?这里怎么又可以任意指定字段进行排序了呢?这是因为 OVER 窗口目前并不完善,不过针对 Top N 这样一个经典应用场景,Flink SQL专门用 OVER 聚合做了优化实现。所以只有在 Top N 的应用场景中,OVER 窗口 ORDER BY后才可以指定其它排序字段;而要想实现 Top N,就必须按照上面的格式进行定义,否则 Flink SQL 的优化器将无法正常解析。而且,目前 Table API 中并不支持 ROW_NUMBER()函数,所以也只有 SQL 中这一种通用的 Top N 实现方式

案例 实时统计点击量前3名的用户

public class TopN {
    public static void main(String[] args) throws Exception {
        // 1.创建一个表执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        env.setParallelism(1);
        env.getConfig().setAutoWatermarkInterval(100); // 100毫秒生成一次水位线

        SingleOutputStreamOperator<Event> streamOperator = env.addSource(new ClickSource())
                // 乱序流的WaterMark生成
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2)) // 延迟2秒保证数据正确
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @Override // 时间戳的提取器
                            public long extractTimestamp(Event event, long l) {
                                return event.getTimestamp();
                            }
                        })
                );

        Table clickTable = tableEnv.fromDataStream(streamOperator, $("user"), $("url"), $("timestamp").as("ts")
                , $("et").rowtime());
        // 将表注册到表环境中
        tableEnv.createTemporaryView("clickTable",clickTable);

        /**
         * 第一步  根据每个用户分组求出其访问量  A = select user, count(url) as cnt from clickTable group by user;
         * 第二步 row_number排名 B = select *, row_number() over ( order by cnt DESC) as rank_num from ( A );
         * 第三步 where过滤 select user, cnt, rank_num from B where rank_num <= 3;
         */
        Table table = tableEnv.sqlQuery("select user, cnt, rank_num from " +
                " ( select *, row_number() over ( order by cnt DESC ) as rank_num from ( select user, count(url) as cnt from clickTable group by user ))" +
                " where rank_num <= 3");

        tableEnv.toChangelogStream(table).print();

        env.execute();
    }
}
6.2 窗口Top N
public class WindowTopN {
    public static void main(String[] args) throws Exception {
        // 1.创建一个表执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        env.setParallelism(1);
        env.getConfig().setAutoWatermarkInterval(100); // 100毫秒生成一次水位线

        SingleOutputStreamOperator<Event> streamOperator = env.addSource(new ClickSource())
                // 乱序流的WaterMark生成
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2)) // 延迟2秒保证数据正确
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @Override // 时间戳的提取器
                            public long extractTimestamp(Event event, long l) {
                                return event.getTimestamp();
                            }
                        })
                );

        Table clickTable = tableEnv.fromDataStream(streamOperator, $("user"), $("url"), $("timestamp").as("ts")
                , $("et").rowtime());
        // 将表注册到表环境中
        tableEnv.createTemporaryView("clickTable",clickTable);

        /**
         * 第一步  根据每个用户分组求出其访问量 并且设置窗口  A = select user, count(url) as cnt, window_start, window_end from table( tumble(table clickTable,DESCRIPTOR(et), INTERVAL '10' SECOND) )  group by user, window_start, window_end;
         * 第二步 row_number排名 并根据窗口分组  B = select *, row_number() over ( partition by window_start, window_end order by cnt DESC) as rank_num from ( A );
         * 第三步 where过滤 select user, cnt, rank_num from B where rank_num <= 3;
         */

        String subQuery = "select user, count(url) as cnt, window_start, window_end " +
                "from table( tumble(table clickTable,DESCRIPTOR(et), INTERVAL '10' SECOND) )  " +
                "group by user, window_start, window_end";

        Table table = tableEnv.sqlQuery("select user, cnt, rank_num from ( " +
                "select *, row_number() over ( partition by window_start, window_end order by cnt DESC) as rank_num " +
                "from ( " + subQuery + " ) ) " +
                "where rank_num <= 3");

        tableEnv.toChangelogStream(table).print();

        env.execute();
    }
}
7.Join

与标准 SQL 一致,Flink SQL 的常规联结也可以分为内联结(INNER JOIN)和外联结(OUTER JOIN),区别在于结果中是否包含不符合联结条件的行。目前仅支持“等值条件”作为联结条件,也就是关键字 ON 后面必须是判断两表中字段相等的逻辑表达式

1. 等值内联结(INNER Equi-JOIN)

内联结用 INNER JOIN 来定义,会返回两表中符合联接条件的所有行的组合,也就是所谓的笛卡尔积(Cartesian product)。目前仅支持等值联结条件

SELECT *
FROM Order
INNER JOIN Product
ON Order.product_id = Product.id

2. 等值外联结(OUTER Equi-JOIN)

与内联结类似,外联结也会返回符合联结条件的所有行的笛卡尔积;另外,还可以将某一侧表中找不到任何匹配的行也单独返回。Flink SQL 支持左外(LEFT JOIN)、右外(RIGHT JOIN)和全外(FULL OUTER JOIN)

这部分知识与标准 SQL 中是完全一样的,这里不再赘述

3.间隔联结查询

我们曾经学习过 DataStream API 中的双流 Join,包括窗口联结(window join)和间隔联结(interval join)。两条流的 Join 就对应着 SQL 中两个表的 Join,这是流处理中特有的联结方式。目前 Flink SQL 还不支持窗口联结,而间隔联结则已经实现

间隔联结(Interval Join)返回的,同样是符合约束条件的两条中数据的笛卡尔积。只不过这里的“约束条件”除了常规的联结条件外,还多了一个时间间隔的限制。具体语法有以下要点:

间隔联结不需要用 JOIN 关键字,直接在 FROM 后将要联结的两表列出来就可以,用逗号分隔。这与标准 SQL 中的语法一致,表示一个“交叉联结”(Cross Join),会返回两表中所有行的笛卡尔积

联结条件用 WHERE 子句来定义,用一个等值表达式描述。交叉联结之后再用 WHERE进行条件筛选,效果跟内联结 INNER JOIN ... ON ...非常类似,我们可以在 WHERE 子句中,联结条件后用 AND 追加一个时间间隔的限制条件;做法是提取左右两侧表中的时间字段,然后用一个表达式来指明两者需要满足的间隔限制。具体定义方式有下面三种,这里分别用 ltime 和 rtime 表示左右表中的时间字段

(1)ltime = rtime
(2)ltime >= rtime AND ltime < rtime + INTERVAL ‘10’ MINUTE
(3)ltime BETWEEN rtime - INTERVAL ‘10’ SECOND AND rtime + INTERVAL ‘5’ SECOND

例如,我们现在除了订单表 Order 外,还有一个“发货表”Shipment,要求在收到订单后四个小时内发货。那么我们就可以用一个间隔联结查询,把所有订单与它对应的发货信息连接合并在一起返回

SELECT *
FROM Order o, Shipment s
WHERE o.id = s.order_id
AND o.order_time BETWEEN s.ship_time - INTERVAL '4' HOUR AND s.ship_time

三.函数

在 SQL 中,我们可以把一些数据的转换操作包装起来,嵌入到 SQL 查询中统一调用,这就是“函数”(functions)。Flink 的 Table API 和 SQL 同样提供了函数的功能。两者在调用时略有不同:Table API 中的函数是通过数据对象的方法调用来实现的;而 SQL 则是直接引用函数名称,传入数据作为参数

由于 Table API 是内嵌在 Java 语言中的,很多方法需要在类中额外添加,因此扩展功能比较麻烦,目前支持的函数比较少;而且 Table API 也不如 SQL 的通用性强,所以一般情况下较少使用。下面我们主要介绍 Flink SQL 中函数的使用

3.1 系统函数

系统函数(System Functions)也叫内置函数(Built-in Functions),是在系统中预先实现好的功能模块。我们可以通过固定的函数名直接调用,实现想要的转换操作。Flink SQL 提供了大量的系统函数,几乎支持所有的标准 SQL 中的操作,这为我们使用 SQL 编写流处理程序提供了极大的方便。Flink SQL 中的系统函数又主要可以分为两大类:标量函数(Scalar Functions)和聚合函数(Aggregate Functions)

标量函数

所谓的“标量”,是指只有数值大小、没有方向的量;所以标量函数指的就是只对输入数据做转换操作、返回一个值的函数。这里的输入数据对应在表中,一般就是一行数据中 1 个或多个字段,因此这种操作有点像流处理转换算子中的 map。另外,对于一些没有输入参数、直接可以得到唯一结果的函数,也属于标量函数

标量函数是最常见、也最简单的一类系统函数,数量非常庞大,很多在标准 SQL 中也有定义。所以我们这里只对一些常见类型列举部分函数,做一个简单概述,具体应用可以查看官网的完整函数列表

  • 比较函数(Comparison Functions)
    比较函数其实就是一个比较表达式,用来判断两个值之间的关系,返回一个布尔类型的值。这个比较表达式可以是用 <、>、= 等符号连接两个值,也可以是用关键字定义的某种判断
    例如:
    (1)value1 = value2 判断两个值相等;
    (2)value1 <> value2 判断两个值不相等
    (3)value IS NOT NULL 判断 value 不为空
  • 逻辑函数(Logical Functions)
    逻辑函数就是一个逻辑表达式,也就是用与(AND)、或(OR)、非(NOT)将布尔类型的值连接起来,也可以用判断语句(IS、IS NOT)进行真值判断;返回的还是一个布尔类型的值
    例如:
    (1)boolean1 OR boolean2 布尔值 boolean1 与布尔值 boolean2 取逻辑或
    (2)boolean IS FALSE 判断布尔值 boolean 是否为 false
    (3)NOT boolean 布尔值 boolean 取逻辑非
  • 算术函数(Arithmetic Functions)
    进行算术计算的函数,包括用算术符号连接的运算,和复杂的数学运算
    例如:
    (1)numeric1 + numeric2 两数相加
    (2)POWER(numeric1, numeric2) 幂运算,取数 numeric1 的 numeric2 次方
    (3)RAND() 返回(0.0, 1.0)区间内的一个 double 类型的伪随机数
  • 字符串函数(String Functions)进行字符串处理的函数
    例如:
    (1)string1 || string2 两个字符串的连接
    (2)UPPER(string) 将字符串 string 转为全部大写
    (3)CHAR_LENGTH(string) 计算字符串 string 的长度
  • 时间函数(Temporal Functions)进行与时间相关操作的函数
    例如:
    (1)DATE string 按格式"yyyy-MM-dd"解析字符串 string,返回类型为 SQL Date
    (2)TIMESTAMP string 按格式"yyyy-MM-dd HH:mm:ss[.SSS]"解析,返回类型为 SQL timestamp
    (3)CURRENT_TIME 返回本地时区的当前时间,类型为 SQL time(与 LOCALTIME等价)
    (4)INTERVAL string range 返回一个时间间隔。string 表示数值;range 可以是 DAY,MINUTE,DAT TO HOUR 等单位,也可以是 YEAR TO MONTH 这样的复合单位。如“2 年10 个月”可以写成:INTERVAL ‘2-10’ YEAR TO MONTH

2. 聚合函数(Aggregate Functions)

聚合函数是以表中多个行作为输入,提取字段进行聚合操作的函数,会将唯一的聚合值作为结果返回。聚合函数应用非常广泛,不论分组聚合、窗口聚合还是开窗(Over)聚合,对数据的聚合操作都可以用相同的函数来定义。
标准 SQL 中常见的聚合函数 Flink SQL 都是支持的,目前也在不断扩展,为流处理应用提供更强大的功能

例如:

  • COUNT(*) 返回所有行的数量,统计个数
  • SUM([ ALL | DISTINCT ] expression) 对某个字段进行求和操作。默认情况下省略了关键字 ALL,表示对所有行求和;如果指定 DISTINCT,则会对数据进行去重,每个值只叠加一次
  • RANK() 返回当前值在一组值中的排名
  • ROW_NUMBER() 对一组值排序后,返回当前值的行号
3.2 UDF

系统函数尽管庞大,也不可能涵盖所有的功能;如果有系统函数不支持的需求,我们就需要用自定义函数(User Defined Functions,UDF)来实现了

Flink 的 Table API 和 SQL 提供了多种自定义函数的接口,以抽象类的形式定义。当前 UDF主要有以下几类:

  • 标量函数(Scalar Functions):将输入的标量值转换成一个新的标量值
  • 表函数(Table Functions):将标量值转换成一个或多个新的行数据,也就是扩展成一个表
  • 聚合函数(Aggregate Functions):将多行数据里的标量值转换成一个新的标量值
  • 表聚合函数(Table Aggregate Functions):将多行数据里的标量值转换成一个或多个新的行数据

(1)UDF - 标量函数 - 1 对 1

自定义标量函数可以把 0 个、 1 个或多个标量值转换成一个标量值,它对应的输入是一行数据中的字段,输出则是唯一的值。所以从输入和输出表中行数据的对应关系看,标量函数是“一对一”的转换

想要实现自定义的标量函数,需要自定义一个类来继承抽象类 ScalarFunction,并实现叫作 eval() 的求值方法。标量函数的行为就取决于求值方法的定义,它必须是公有的(public),而且名字必须是 eval。求值方法 eval 可以重载多次,任何数据类型都可作为求值方法的参数和返回值类型,写完后将类注册到表环境就可以直接在SQL 中调用了

这里需要特别说明的是,ScalarFunction 抽象类中并没有定义 eval()方法,所以我们不能直接在代码中重写(override);但 Table API 的框架底层又要求了求值方法必须名字为 eval()。这是 Table API 和 SQL 目前还显得不够完善的地方,未来的版本应该会有所改进。ScalarFunction 以及其它所有的 UDF 接口,都在 org.apache.flink.table.functions 中

案例 自定义标量函数 来判断数据时第几个

public class ScalarFunctionsTest {
    public static void main(String[] args) throws Exception {
        // 创建一个表执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        env.setParallelism(1);
        env.getConfig().setAutoWatermarkInterval(100); // 100毫秒生成一次水位线

        SingleOutputStreamOperator<Event> streamOperator = env.addSource(new ClickSource())
                // 乱序流的WaterMark生成
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2)) // 延迟2秒保证数据正确
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @Override // 时间戳的提取器
                            public long extractTimestamp(Event event, long l) {
                                return event.getTimestamp();
                            }
                        })
                );

        Table clickTable = tableEnv.fromDataStream(streamOperator, $("user"), $("url"), $("timestamp").as("ts")
                , $("et").rowtime()); // et是事件时间
        // 将表注册到表环境中
        tableEnv.createTemporaryView("clickTable",clickTable);

        // 注册自定义的函数
        tableEnv.createTemporarySystemFunction("MyScalarFunction", MyScalarFunction.class);

        Table table = tableEnv.sqlQuery("select user, MyScalarFunction(user) from clickTable");

        tableEnv.toDataStream(table).print();

        env.execute();
    }

    public static class MyScalarFunction extends ScalarFunction {
        int count;
        public int eval(String name){
            return ++count;
        }
    }
}

flinksql 连接数据库超时是什么原因 flink1.11 sql,flinksql 连接数据库超时是什么原因 flink1.11 sql_SQL_07,第7张

(2)UDF - 表函数 - 一对多

我们使用表函数,可以对一行数据得到一个表

要实现自定义的表函数,需要自定义类来继承抽象类 TableFunction,内部必须要实现的也是一个名为 eval 的求值方法。与标量函数不同的是,TableFunction 类本身是有一个泛型参数T 的,这就是表函数返回数据的类型;而 eval()方法没有返回类型,内部也没有 return语句,是通过调用 collect()方法来发送想要输出的行数据的

我们使用表函数,可以对一行数据得到一个表,这和 Hive 中的 UDTF 非常相似。那对于原先输入的整张表来说,又该得到什么呢?一个简单的想法是,就让输入表中的每一行,与它转换得到的表进行联结(join),然后再拼成一个完整的大表,这就相当于对原来的表进行了扩展。在 Hive 的 SQL 语法中,提供了“侧向视图”(lateral view,也叫横向视图)的功能,可以将表中的一行数据拆分成多行;Flink SQL 也有类似的功能,是用 LATERAL TABLE 语法来实现的

在 SQL 中调用表函数,需要使用 LATERAL TABLE(<TableFunction>)来生成扩展的“侧向表”,然后与原始表进行联结(Join)。这里的 Join 操作可以是直接做交叉联结(cross join),在 FROM 后用逗号分隔两个表就可以;也可以是以 ON TRUE 为条件的左联结(LEFT JOIN)

public class TableFunctionTest {
    public static void main(String[] args) throws Exception {
        // 创建一个表执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        env.setParallelism(1);
        env.getConfig().setAutoWatermarkInterval(100); // 100毫秒生成一次水位线

        SingleOutputStreamOperator<Event> streamOperator = env.addSource(new ClickSource())
                // 乱序流的WaterMark生成
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2)) // 延迟2秒保证数据正确
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @Override // 时间戳的提取器
                            public long extractTimestamp(Event event, long l) {
                                return event.getTimestamp();
                            }
                        })
                );

        Table clickTable = tableEnv.fromDataStream(streamOperator, $("user"), $("url"), $("timestamp").as("ts")
                , $("et").rowtime()); // et是事件时间
        // 将表注册到表环境中
        tableEnv.createTemporaryView("clickTable",clickTable);

        // 注册自定义表函数
        tableEnv.createTemporaryFunction("MyTableFunction",MyTableFunction.class);

		// as T 起别名
        Table table = tableEnv.sqlQuery("select user, url, word, length " +
                "from clickTable, LATERAL TABLE( MyTableFunction(url) ) as T(word,length)");

        tableEnv.toDataStream(table).print();

        env.execute();


    }

    public static class MyTableFunction extends TableFunction<Tuple2<String,Integer>>{
        // 我们将url以问号分割 分成多行 每一行是参数 + 参数的长度
        // 例如 ./home?name=10 将被分成 ./home 6 和 name = 10 7 两行
        public void eval(String str){
            String[] split = str.split("\?");
            for (String s : split) {
                collect(Tuple2.of(s,s.length()));
            }
        }
    }
}

flinksql 连接数据库超时是什么原因 flink1.11 sql,flinksql 连接数据库超时是什么原因 flink1.11 sql_字段_08,第8张

(3)UDF - 聚合函数 - 多对一

用户自定义聚合函数(User Defined AGGregate function,UDAGG)会把一行或多行数据(也就是一个表)聚合成一个标量值。这是一个标准的“多对一”的转换

自定义聚合函数需要继承抽象类 AggregateFunction。AggregateFunction 有两个泛型参数<T, ACC>,T 表示聚合输出的结果类型,ACC 则表示聚合的中间状态类型,每个 AggregateFunction 都必须实现以下几个方法:

  • createAccumulator()
    这是创建累加器的方法。没有输入参数,返回类型为累加器类型 ACC
  • accumulate()
    这是进行聚合计算的核心方法,每来一行数据都会调用。它的第一个参数是确定的,就是当前的累加器,类型为 ACC,表示当前聚合的中间状态;后面的参数则是聚合函数调用时传入的参数,可以有多个,类型也可以不同。这个方法主要是更新聚合状态,所以没有返回类型
  • getValue()
    这是得到最终返回结果的方法。输入参数是 ACC 类型的累加器,输出类型为 T。在遇到复杂类型时,Flink 的类型推导可能会无法得到正确的结果。所以AggregateFunction也可以专门对累加器和返回结果的类型进行声明,这是通过 getAccumulatorType()和getResultType()两个方法来指定的

除了上面的方法,还有几个方法是可选的。这些方法有些可以让查询更加高效,有些是在某些特定场景下必须要实现的。比如,如果是对会话窗口进行聚合,merge()方法就是必须要实现的,它会定义累加器的合并操作,而且这个方法对一些场景的优化也很有用;而如果聚合函数用在 OVER 窗口聚合中,就必须实现 retract()方法,保证数据可以进行撤回操作;resetAccumulator()方法则是重置累加器,这在一些批处理场景中会比较有用。AggregateFunction 的所有方法都必须是 公有的(public),不能是静态的(static),而且名字必须跟上面写的完全一样。 createAccumulator 、 getValue 、 getResultType 以 及getAccumulatorType 这几个方法是在抽象类 AggregateFunction 中定义的,可以 override;而其他则都是底层架构约定的方法

(4)表聚合函数(Table Aggregate Functions)

用户自定义表聚合函数(UDTAGG)可以把一行或多行数据(也就是一个表)聚合成另一张表,结果表中可以有多行多列。很明显,这就像表函数和聚合函数的结合体,是一个“多对多”的转换

自定义表聚合函数需要继承抽象类 TableAggregateFunction。TableAggregateFunction 的结构和原理与 AggregateFunction 非常类似,同样有两个泛型参数<T, ACC>,用一个 ACC 类型的累加器(accumulator)来存储聚合的中间结果。聚合函数中必须实现的三个方法,在TableAggregateFunction 中也必须对应实现:

  • createAccumulator()
    创建累加器的方法,与 AggregateFunction 中用法相同
  • accumulate()
    聚合计算的核心方法,与 AggregateFunction 中用法相同
  • emitValue()
    所有输入行处理完成后,输出最终计算结果的方法。这个方法对应着 AggregateFunction中的 getValue()方法;区别在于 emitValue 没有输出类型,而输入参数有两个:第一个是 ACC类型的累加器,第二个则是用于输出数据的“收集器”out,它的类型为 Collect。所以很明显,表聚合函数输出数据不是直接 return,而是调用 out.collect()方法,调用多次就可以输出多行数据了;这一点与表函数非常相似。另外,emitValue()在抽象类中也没有定义,无法 override,必须手动实现

目前 SQL 中没有直接使用表聚合函数的方式,所以需要使用 Table API 的方式来调用

四.SQL 客户端

Flink 为我们提供了一个工具来进行 Flink 程序的编写、测试和提交,这工具叫作“SQL 客户端”。SQL 客户端提供了一个命令行交互界面(CLI),我们可以在里面非常容易地编写 SQL 进行查询,就像使用 MySQL 一样;整个 Flink 应用编写、提交的过程全变成了写 SQL,不需要写一行 Java/Scala 代码

具体使用流程如下:
(1)首先启动本地集群

./bin/start-cluster.sh

flinksql 连接数据库超时是什么原因 flink1.11 sql,flinksql 连接数据库超时是什么原因 flink1.11 sql_flink_09,第9张

(2)启动 Flink SQL 客户端

./bin/sql-client.sh

flinksql 连接数据库超时是什么原因 flink1.11 sql,flinksql 连接数据库超时是什么原因 flink1.11 sql_大数据_10,第10张

SQL 客户端的启动脚本同样位于 Flink 的 bin 目录下。默认的启动模式是 embedded,也就是说客户端是一个嵌入在本地的进程,这是目前唯一支持的模式。未来会支持连接到远程 SQL客户端的模式

(3)设置运行模式
启动客户端后,就进入了命令行界面,这时就可以开始写 SQL 了。一般我们会在开始之前对环境做一些设置,比较重要的就是运行模式

首先是表环境的运行时模式,有流处理和批处理两个选项。默认为流处理:

Flink SQL> SET 'execution.runtime-mode' = 'streaming';

其次是 SQL 客户端的“执行结果模式”,主要有 table、changelog、tableau 三种,默认为table 模式:

Flink SQL> SET 'sql-client.execution.result-mode' = 'table';

table 模式就是最普通的表处理模式,结果会以逗号分隔每个字段;changelog 则是更新日志模式,会在数据前加上“+”(表示插入)或“-”(表示撤回)的前缀;而 tableau 则是经典的可视化表模式,结果会是一个虚线框的表格

此外我们还可以做一些其它可选的设置,比如之前提到的空闲状态生存时间(TTL):

Flink SQL> SET 'table.exec.state.ttl' = '1000';

除了在命令行进行设置,我们也可以直接在 SQL 客户端的配置文件 sql-cli-defaults.yaml中进行各种配置,甚至还可以在这个 yaml 文件里预定义表、函数和 catalog。关于配置文件的更多用法,可以查阅官网的详细说明

(4)执行 SQL 查询

Flink SQL> CREATE TABLE EventTable(
> user STRING,
> url STRING,
> `timestamp` BIGINT
> ) WITH (
> 'connector' = 'filesystem',
> 'path' = 'events.csv',
> 'format' = 'csv'
> );
Flink SQL> CREATE TABLE ResultTable (
> user STRING,
> cnt BIGINT
> ) WITH (
> 'connector' = 'print'
> );
Flink SQL> INSERT INTO ResultTable SELECT user, COUNT(url) as cnt FROM EventTable
GROUP BY user;

五.连接外部系统

1.文件系统

Flink 提供了文件系统的连接器,支持从本地或者分布式的文件系统中读写数据。这个连接器是内置在 Flink 中的,所以使用它并不需要额外引入依赖

下面是一个连接到文件系统的示例:

CREATE TABLE MyTable (
 column_name1 INT,
 column_name2 STRING,
 ...
 part_name1 INT,
 part_name2 STRING
) PARTITIONED BY (part_name1, part_name2) WITH (
 'connector' = 'filesystem', -- 连接器类型
 'path' = '...', -- 文件路径
 'format' = '...' -- 文件格式
)

这里在 WITH 前使用了 PARTITIONED BY 对数据进行了分区操作。文件系统连接器支持对分区文件的访问

2.Kafka

想要在 Flink 程序中使用 Kafka 连接器,需要引入如下依赖:

<dependency>
 <groupId>org.apache.flink</groupId>
 <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
 <version>${flink.version}</version>
</dependency>

(1)kafka作为source

CREATE TABLE KafkaTable (
`user` STRING,
 `url` STRING,
 `ts` TIMESTAMP(3) METADATA FROM 'timestamp'
) WITH (
 'connector' = 'kafka',
 'topic' = 'events',
 'properties.bootstrap.servers' = 'localhost:9092',
 'properties.group.id' = 'testGroup',
 'scan.startup.mode' = 'earliest-offset',
 'format' = 'csv'
)

这里定义了 Kafka 连接器对应的主题(topic),Kafka 服务器,消费者组 ID,消费者起始模式以及表格式。需要特别说明的是,在 KafkaTable 的字段中有一个 ts,它的声明中用到了METADATA FROM,这是表示一个“元数据列”(metadata column),它是由 Kafka 连接器的元数据“timestamp”生成的。这里的 timestamp 其实就是 Kafka 中数据自带的时间戳,我们把它直接作为元数据提取出来,转换成一个新的字段 ts

(2)kafka作为 sink

正常情况下,Kafka 作为保持数据顺序的消息队列,读取和写入都应该是流式的数据,对应在表中就是仅追加(append-only)模式。如果我们想要将有更新操作(比如分组聚合)的结果表写入 Kafka,就会因为 Kafka 无法识别撤回(retract)或更新插入(upsert)消息而导致异常

为了解决这个问题,Flink 专门增加了一个“更新插入 Kafka”(Upsert Kafka)连接器。这个连接器支持以更新插入(UPSERT)的方式向 Kafka 的 topic 中读写数据。具体来说,Upsert Kafka 连接器处理的是更新日志(changlog)流。如果作为 TableSource,连接器会将读取到的 topic中的数据(key, value),解释为对当前 key 的数据值的更新(UPDATE),也就是查找动态表中 key 对应的一行数据,将 value 更新为最新的值;因为是 Upsert 操作,所以如果没有 key 对应的行,那么也会执行插入(INSERT)操作。另外,如果遇到 value 为空
(null),连接器就把这条数据理解为对相应 key 那一行的删除(DELETE)操作

如果作为 TableSink,Upsert Kafka 连接器会将有更新操作的结果表,转换成更新日志(changelog)流。如果遇到插入(INSERT)或者更新后(UPDATE_AFTER)的数据,对应的是一个添加(add)消息,那么就直接正常写入 Kafka 主题;如果是删除(DELETE)或者更新前的数据,对应是一个撤回(retract)消息,那么就把 value 为空(null)的数据写入 Kafka。由于 Flink 是根据键(key)的值对数据进行分区的,这样就可以保证同一个 key 上的更新和删除消息都会落到同一个分区中

下面是一个创建和使用 Upsert Kafka 表的例子:

CREATE TABLE pageviews_per_region (
 user_region STRING,
 pv BIGINT,
 uv BIGINT,
 PRIMARY KEY (user_region) NOT ENFORCED
) WITH (
 'connector' = 'upsert-kafka',
 'topic' = 'pageviews_per_region',
 'properties.bootstrap.servers' = '...',
 'key.format' = 'avro',
 'value.format' = 'avro'
);
CREATE TABLE pageviews (
 user_id BIGINT,
 page_id BIGINT,
 viewtime TIMESTAMP,
 user_region STRING,
 WATERMARK FOR viewtime AS viewtime - INTERVAL '2' SECOND
) WITH (
 'connector' = 'kafka',
 'topic' = 'pageviews',
 'properties.bootstrap.servers' = '...',
 'format' = 'json'
);
-- 计算 pv、uv 并插入到 upsert-kafka 表中
INSERT INTO pageviews_per_region
SELECT
 user_region,
 COUNT(*),
 COUNT(DISTINCT user_id)
FROM pageviews
GROUP BY user_region;

这里我们从 Kafka 表 pageviews 中读取数据,统计每个区域的 PV(全部浏览量)和 UV
(对用户去重),这是一个分组聚合的更新查询,得到的结果表会不停地更新数据。为了将结
果表写入 Kafka 的 pageviews_per_region 主题,我们定义了一个 Upsert Kafka 表,它的字段中
需要用PRIMARY KEY来指定主键,并且在WITH子句中分别指定key和value的序列化格式

3.JDBC

关系型数据表本身就是 SQL 最初应用的地方,所以我们也会希望能直接向关系型数据库中读写表数据。Flink 提供的 JDBC 连接器可以通过 JDBC 驱动程序(driver)向任意的关系型数据库读写数据,比如 MySQL、PostgreSQL、Derby 等

作为 TableSink 向数据库写入数据时,运行的模式取决于创建表的 DDL 是否定义了主键(primary key)。如果有主键,那么 JDBC 连接器就将以更新插入(Upsert)模式运行,可以向外部数据库发送按照指定键(key)的更新(UPDATE)和删除(DELETE)操作;如果没有定义主键,那么就将在追加(Append)模式下运行,不支持更新和删除操作

1. 引入依赖

想要在 Flink 程序中使用 JDBC 连接器,需要引入如下依赖:

<dependency>
	<groupId>org.apache.flink</groupId>
	<artifactId>flink-connector-jdbc_${scala.binary.version}</artifactId>
	<version>${flink.version}</version>
</dependency>

此外,为了连接到特定的数据库,我们还用引入相关的驱动器依赖,比如 MySQL:

<dependency>
 <groupId>mysql</groupId>
 <artifactId>mysql-connector-java</artifactId>
 <version>5.1.38</version>
</dependency>

创建 JDBC 表的方法与前面 Upsert Kafka 大同小异。下面是一个具体示例:

-- 创建一张连接到 MySQL 的 表
CREATE TABLE MyTable (
 id BIGINT,
 name STRING,
 age INT,
 status BOOLEAN,
 PRIMARY KEY (id) NOT ENFORCED
) WITH (
 'connector' = 'jdbc',
 'url' = 'jdbc:mysql://localhost:3306/mydatabase',
 'table-name' = 'users'
);

– 将另一张表 T 的数据写入到 MyTable 表中

INSERT INTO MyTable
SELECT id, name, age, status FROM T;

这里创建表的 DDL 中定义了主键,所以数据会以 Upsert 模式写入到 MySQL 表中;而到MySQL 的连接,是通过 WITH 子句中的 url 定义的。要注意写入 MySQL 中真正的表名称是users,而 MyTable 是注册在 Flink 表环境中的表

4.Elasticsearch

Elasticsearch 作为分布式搜索分析引擎,在大数据应用中有非常多的场景。Flink 提供的Elasticsearch的SQL连接器只能作为TableSink,可以将表数据写入Elasticsearch的索引(index)

Elasticsearch 连接器的使用与 JDBC 连接器非常相似,写入数据的模式同样是由创建表的 DDL
中是否有主键定义决定的

1. 引入依赖

想要在 Flink 程序中使用 Elasticsearch 连接器,需要引入对应的依赖。具体的依赖与Elasticsearch 服务器的版本有关,对于 6.x 版本引入依赖如下:

<dependency>
 <groupId>org.apache.flink</groupId> 
<artifactId>flink-connector-elasticsearch6_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>
// 对于 Elasticsearch 7 以上的版本,引入的依赖则是:
<dependency>
 <groupId>org.apache.flink</groupId> 
<artifactId>flink-connector-elasticsearch7_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>

2. 创建连接到 Elasticsearch 的表

创建 Elasticsearch 表的方法与 JDBC 表基本一致。下面是一个具体示例:

-- 创建一张连接到 Elasticsearch 的 表
CREATE TABLE MyTable (
 user_id STRING,
 user_name STRING
 uv BIGINT,
 pv BIGINT,
 PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
 'connector' = 'elasticsearch-7',
 'hosts' = 'http://localhost:9200',
 'index' = 'users'
);

这里定义了主键,所以会以更新插入(Upsert)模式向 Elasticsearch 写入数据

5.Hbase

在流处理场景下,连接器作为 TableSink 向 HBase 写入数据时,采用的始终是更新插入(Upsert)模式。也就是说,HBase 要求连接器必须通过定义的主键(primary key)来发送更新日志(changelog)。所以在创建表的 DDL 中,我们必须要定义行键(rowkey)字段,并将它声明为主键;如果没有用 PRIMARY KEY 子句声明主键,连接器会默认把 rowkey 作为主键

1. 引入依赖
想要在 Flink 程序中使用 HBase 连接器,需要引入对应的依赖。目前 Flink 只对 HBase 的1.4.x 和 2.2.x 版本提供了连接器支持,而引入的依赖也应该与具体的 HBase 版本有关。对于1.4 版本引入依赖如下:

<dependency>
 <groupId>org.apache.flink</groupId>
 <artifactId>flink-connector-hbase-1.4_${scala.binary.version}</artifactId>
 <version>${flink.version}</version>
</dependency>

对于 HBase 2.2 版本,引入的依赖则是:

<dependency>
 <groupId>org.apache.flink</groupId>
 <artifactId>flink-connector-hbase-2.2_${scala.binary.version}</artifactId>
 <version>${flink.version}</version>
</dependency>

2. 创建连接到 HBase 的表

由于 HBase 并不是关系型数据库,因此转换为 Flink SQL 中的表会稍有一些麻烦。在 DDL创建出的 HBase 表中,所有的列族(column family)都必须声明为 ROW 类型,在表中占据一个字段;而每个 family 中的列(column qualifier)则对应着 ROW 里的嵌套字段。我们不需要将 HBase 中所有的 family 和 qualifier 都在 Flink SQL 的表中声明出来,只要把那些在查询中用到的声明出来就可以了

除了所有 ROW 类型的字段(对应着 HBase 中的 family),表中还应有一个原子类型的字段,它就会被识别为 HBase 的 rowkey。在表中这个字段可以任意取名,不一定非要叫 rowkey。下面是一个具体示例:

-- 创建一张连接到 HBase 的 表
CREATE TABLE MyTable (
rowkey INT,
family1 ROW<q1 INT>,
family2 ROW<q2 STRING, q3 BIGINT>,
family3 ROW<q4 DOUBLE, q5 BOOLEAN, q6 STRING>,
PRIMARY KEY (rowkey) NOT ENFORCED
) WITH (
'connector' = 'hbase-1.4',
'table-name' = 'mytable',
'zookeeper.quorum' = 'localhost:2181'
);

– 假设表 T 的字段结构是 [rowkey, f1q1, f2q2, f2q3, f3q4, f3q5, f3q6]

INSERT INTO MyTable
SELECT rowkey, ROW(f1q1), ROW(f2q2, f2q3), ROW(f3q4, f3q5, f3q6) FROM T;

我们将另一张 T 中的数据提取出来,并用 ROW()函数来构造出对应的 column family,最终写入 HBase 中名为 mytable 的表

6.Hive

Flink 与 Hive 的集成比较特别。Flink 提供了“Hive 目录”(HiveCatalog)功能,允许使用Hive 的“元存储”(Metastore)来管理 Flink 的元数据。这带来的好处体现在两个方面:
(1)Metastore 可以作为一个持久化的目录,因此使用 HiveCatalog 可以跨会话存储 Flink特定的元数据。这样一来,我们在 HiveCatalog 中执行执行创建 Kafka 表或者 ElasticSearch 表,就可以把它们的元数据持久化存储在 Hive 的 Metastore 中;对于不同的作业会话就不需要重复创建了,直接在 SQL 查询中重用就可以

(2)使用 HiveCatalog,Flink 可以作为读写 Hive 表的替代分析引擎。这样一来,在 Hive中进行批处理会更加高效;与此同时,也有了连续在 Hive 中读写数据、进行流处理的能力,这也使得“实时数仓”(real-time data warehouse)成为了可能。HiveCatalog 被设计为“开箱即用”,与现有的 Hive 配置完全兼容,我们不需要做任何的
修改与调整就可以直接使用。注意只有 Blink 的计划器(planner)提供了 Hive 集成的支持,所以需要在使用 Flink SQL时选择Blink planner。下面我们就来看以下与Hive 集成的具体步骤

1. 引入依赖

由于 Hive 是基于 Hadoop 的组件,因此我们首先需要提供 Hadoop 的相关支持,在环境变量中设置

HADOOP_CLASSPATH:
export HADOOP_CLASSPATH=`hadoop classpath`

在 Flink 程序中可以引入以下依赖:

<dependency>
	 <groupId>org.apache.flink</groupId>
	 <artifactId>flink-connector-hive_${scala.binary.version}</artifactId>
	 <version>${flink.version}</version>
</dependency>
<!-- Hive 依赖 -->
<dependency>
	<groupId>org.apache.hive</groupId>
	<artifactId>hive-exec</artifactId>
	<version>${hive.version}</version>
</dependency>

建议不要把这些依赖打包到结果 jar 文件中,而是在运行时的集群环境中为不同的 Hive版本添加不同的依赖支持。具体版本对应的依赖关系,可以查询官网说明

2. 连接到 Hive

在 Flink 中连接 Hive,是通过在表环境中配置 HiveCatalog 来实现的。需要说明的是,配置 HiveCatalog 本身并不需要限定使用哪个 planner,不过对 Hive 表的读写操作只有 Blink 的planner 才支持。所以一般我们需要将表环境的 planner 设置为 Blink

下面是代码中配置 Catalog 的示例:

EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().build();
TableEnvironment tableEnv = TableEnvironment.create(settings);
String name = "myhive";
String defaultDatabase = "mydatabase";
String hiveConfDir = "/opt/hive-conf";
// 创建一个 HiveCatalog,并在表环境中注册
HiveCatalog hive = new HiveCatalog(name, defaultDatabase, hiveConfDir);
tableEnv.registerCatalog("myhive", hive);
// 使用 HiveCatalog 作为当前会话的 catalog
tableEnv.useCatalog("myhive");

3.设置 SQL 方言

我们知道,Hive内部提供了类SQL的查询语言,不过语法细节与标准SQL会有一些出入,相当于是 SQL 的一种“方言”(dialect)。为了提高与 Hive 集成时的兼容性,Flink SQL 提供了一个非常有趣而强大的功能:可以使用方言来编写 SQL 语句。换句话说,我们可以直接在 Flink中写 Hive SQL 来操作 Hive 表,这无疑给我们的读写处理带来了极大的方便

Flink 目前支持两种 SQL 方言的配置:default 和 hive。所谓的 default 就是 Flink SQL 默认的 SQL 语法了。我们需要先切换到 hive 方言,然后才能使用 Hive SQL 的语法。具体设置可以分为 SQL 和 Table API 两种方式

(1)SQL 中设置
我们可以通过配置 table.sql-dialect 属性来设置 SQL 方言:
set table.sql-dialect=hive;

(2)Table API 中设置
另外一种方式就是在代码中,直接使用 Table API 获取表环境的配置项来进行设置:

// 配置 hive 方言
tableEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
// 配置 default 方言
tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);

4.读写 Hive 表

有了 SQL 方言的设置,我们就可以很方便的在 Flink 中创建 Hive 表并进行读写操作了。Flink 支持以批处理和流处理模式向 Hive 中读写数据。在批处理模式下,Flink 会在执行查询语句时对 Hive 表进行一次性读取,在作业完成时将结果数据向 Hive 表进行一次性写入;而在流处理模式下,Flink 会持续监控 Hive 表,在新数据可用时增量读取,也可以持续写入新数据并增量式地让它们可见

更灵活的是,我们可以随时切换 SQL 方言,从其它数据源(例如 Kafka)读取数据、经转换后再写入Hive。下面是以纯SQL形式编写的一个示例,我们可以启动SQL客户端来运行:

-- 设置 SQL 方言为 hive,创建 Hive 表
SET table.sql-dialect=hive;
CREATE TABLE hive_table (
 user_id STRING,
 order_amount DOUBLE
) PARTITIONED BY (dt STRING, hr STRING) STORED AS parquet TBLPROPERTIES (
 'partition.time-extractor.timestamp-pattern'='$dt $hr:00:00',
 'sink.partition-commit.trigger'='partition-time',
 'sink.partition-commit.delay'='1 h',
 'sink.partition-commit.policy.kind'='metastore,success-file'
);
-- 设置 SQL 方言为 default,创建 Kafka 表
SET table.sql-dialect=default;
CREATE TABLE kafka_table (
 user_id STRING,
 order_amount DOUBLE,
 log_ts TIMESTAMP(3),
 WATERMARK FOR log_ts AS log_ts - INTERVAL '5' SECOND – 定义水位线
) WITH (...);
-- 将 Kafka 中读取的数据经转换后写入 Hive 
INSERT INTO TABLE hive_table 
SELECT user_id, order_amount, DATE_FORMAT(log_ts, 'yyyy-MM-dd'), 
DATE_FORMAT(log_ts, 'HH')
FROM kafka_table;

这里我们创建 Hive 表时设置了通过分区时间来触发提交的策略。将 Kafka 中读取的数据经转换后写入 Hive,这是一个流处理的 Flink SQL 程序



https://www.xamrdz.com/lan/5zj1926332.html

相关文章: