Flink SQL#

Function 内置函数#

FROM_UNIXTIME#

FROM_UNIXTIME(BIGINT)

传入一个十位数转化时间字符串

TO_TIMESTAMP#

FROM_UNIXTIME(BIGINT) 传入一个十三位数转化时间类型
FROM_TIMESTAMP(DATE) 传入时间字符串转化时间类型

WATERMARK#

严格递增时间戳：#

1
WATERMARK FOR rowtime_column AS rowtime_column

发出到目前为止已观察到的最大时间戳的 watermark ，时间戳大于最大时间戳的行被认为没有迟到。

递增时间戳:#

延时 5 秒生成 watermark

1
WATERMARK FOR time_ltz AS time_ltz - INTERVAL '5' SECOND

发出到目前为止已观察到的最大时间戳减 1 的 watermark ，时间戳大于或等于最大时间戳的行被认为没有迟到。

Query 查询#

DISTINCT 去重#

1
SELECT DISTINCT id FROM Orders

WINDOW 窗口#

TUMBLE 滚动窗口#

1
TUMBLE(TABLE data,DESCRIPTOR(timecol), size)

data 表名 e.g. TABLE demo

timecol timestamp 字段 e.g. DESCRIPTOR(bidtime)

size 滚动窗口大小 e.g. INTERVAL '10' MINUTES

e.g.

1
select * from TABLE(
2
    TUMBLE(
3
        TABLE demo,
4
        DESCRIPTOR(bidtime),
5
        INTERVAL '10' MINUTES
6
    )
7
)

1
select * from TABLE(
2
    TUMBLE(
3
        DATA => TABLE demo,
4
        TIMECOL => DESCRIPTOR(bidtime),
5
        SIZE => INTERVAL '10' MINUTES
6
    )
7
)

实际使用:

1
// 实际使用中加上window_start, window_end
2
envTable.sqlQuery(
3
   """
4
    |select `window_start`,`window_end`,`UserId` from TABLE(
5
    |   TUMBLE (
6
    |       TABLE add,
7
    |       DESCRIPTOR(`TimeStamp`),
8
    |       INTERVAL '10' MINUTES
9
    |   )
10
    |)
11
    |where action = 'pv'
12
    |GROUP BY `window_start`, `window_end`,`UserId`
13
    |""".stripMargin).limit(1)

HOP 滑动窗口#

1
HOP(TABLE data, DESCRIPTOR(timecol), slide, size [, offset ])

data 表名 e.g. `TABLE demo

timecol timestamp 字段 e.g. DESCRIPTOR(bidtime)

slide 滑动大小 e.g. INTERVAL '5' MINUTES

size 滑动窗口大小 e.g. INTERVAL '10' MINUTES

e.g.

1
select * from TABLE(
2
    HOP(
3
        TABLE demo,
4
        DESCRIPTOR(bidtime),
5
        INTERVAL '5' MINUTES,
6
        INTERVAL '10' MINUTES
7
       )
8
)

1
select * from TABLE(
2
    HOP(
3
        DATA => TABLE demo,
4
        TIMECOL => DESCRIPTOR(bidtime),
5
        SLIDE => INTERVAL '5' MINUTES,
6
        SIZE => INTERVAL '10' MINUTES
7
       )
8
)

Create 创建#

TABLE 表#

行

Insert 插入#

Flink Table API#

Operations 操作API#

from#

1
tableEnv.from('Orders')
2
等同于 select * from Orders

本身等同于sql中 from

FromValues#

1
val table = tableEnv.fromValues(
2
  row(1,"ABC"),
3
    row(2,"ABCDE")
4
)

效果如下：

1
root
2
| -- f0 BIGINT NOT NULL
3
| -- f1 VARCHAR(5) NOT NULL

默认自动识别类型可以指定类型如下:

1
val table = tableEnv.fromValues(
2
  DataTypes.ROW(
3
      DataTypes.FIELD("id",DataTypes.DECIMAL(10,2)),
4
        DataTypes.FIELD("name",DataTypes.STRING())
5
    ),
6
    row(1,"ABC"),
7
    row(2,"ABCDE")
8
)

结构如下:

1
root
2
| -- id DECIMAL(10,2)
3
| -- name STRING

本身等同于sql中 values

Select#

1
val orders = tableEnv.from("Orders")
2
Table result = orders.select($"a", $"c" as "d")
3
// or
4
Table result = orders.select($"*")

本身等同于sql中 select

As#

1
val orders = tableEnv.from("Orders");
2
val result = orders.as("x, y, z, t");

Where / Filter#

1
val orders = tableEnv.from("Orders");
2
val result = orders.where($("b").isEqual("red"));
3
// select * from Orders where b = 'red'
4
// or
5
val orders = tableEnv.from("Orders");
6
val result = orders.filter($("b").isEqual("red"));
7
// select * from Orders where b != 'red'

列操作#

AddColumns#

执行字段添加操作。如果所添加的字段已经存在，将抛出异常。

1
val orders = tableEnv.from("Orders");
2
val result = orders.addColumns(concat($"c"));

AddOrReplaceColumns#

执行字段添加操作。如果添加的列名称和已存在的列名称相同，则已存在的字段将被替换。此外，如果添加的字段里面有重复的字段名，则会使用最后一个字段。

1
val orders = tableEnv.from("Orders")
2
val result = orders.addOrReplaceColumns(concat($"c", "Sunny") as "desc")

DropColumns#

1
val orders = tableEnv.from("Orders")
2
val result = orders.dropColumns($"b")-

RenameColumns#

1
val orders = tableEnv.from("Orders")
2
val result = orders.renameColumns($"b" as "b2")

Aggregations 聚合API#

GroupBy Aggregation 分组聚合#

1
val orders: Table = tableEnv.from("Orders")
2
val result = orders.groupBy($"a").select($"a", $"b".sum().as("d"))

GroupBy Window Aggregation 窗口聚合#

1
val orders: Table = tableEnv.from("Orders")
2
val result: Table = orders
3
    .window(Tumble over 5.minutes on $"rowtime" as "w") // 定义窗口
4
    .groupBy($"a", $"w") // 按窗口和键分组
5
    .select($"a", $"w".start, $"w".end, $"w".rowtime, $"b".sum as "d") // 访问窗口属性并聚合

Distinct Aggregation 去重聚合#

1
val orders: Table = tableEnv.from("Orders")
2
// 按属性分组后的的互异（互不相同、去重）聚合
3
val groupByDistinctResult = orders
4
    .groupBy($"a")
5
    .select($"a", $"b".sum.distinct as "d")
6
// 按属性、时间窗口分组后的互异（互不相同、去重）聚合
7
val groupByWindowDistinctResult = orders
8
    .window(Tumble over 5.minutes on $"rowtime" as "w").groupBy($"a", $"w")
9
    .select($"a", $"b".sum.distinct as "d")
10
// over window 上的互异（互不相同、去重）聚合
11
val result = orders
12
    .window(Over
13
        partitionBy $"a"
14
        orderBy $"rowtime"
15
        preceding UNBOUNDED_RANGE
16
        as $"w")
17
    .select($"a", $"b".avg.distinct over $"w", $"b".max over $"w", $"b".min over $"w")

Connector 连接器#

FileSystem 文件系统#

参数#

path = 'file:///' + 路径

案例#

1
envTable.executeSql(
2
"""
3
        |create table add (
4
        |`UserId` bigint
5
        |) WITH (
6
        |'connector' = 'filesystem',
7
        |'path' = 'file:///D:\project\java\reflink\reflink\source\UserBehavior.csv',
8
        |'format' = 'csv'
9
        |)
10
        |""".stripMargin)

JDBC#

pom.xml

1
<dependency>
2
  <groupId>org.apache.flink</groupId>
3
  <artifactId>flink-connector-jdbc_2.11</artifactId>
4
  <version>1.13.6</version>
5
</dependency>

$ urlJDBC 数据库 url。
table-name 连接到 JDBC 表的名称。
driver 用于连接到此 URL 的 JDBC 驱动类名，如果不设置，将自动从 URL 中推导。
username JDBC 用户名
password JDBC 密码。

注意没有 foramt

案例#

1
envTable.executeSql(
2
"""
3
        |create table add (
4
        |`UserId` bigint
5
        |) WITH (
6
        |'connector' = 'jdbc',
7
        |'url' = 'jdbc:mysql://localhost:3306/test'
8
        |
9
        |)
10
        |""".stripMargin)

Kafka#

pom.xml

1
<dependency>
2
  <groupId>org.apache.flink</groupId>
3
  <artifactId>flink-connector-kafka_2.11</artifactId>
4
  <version>${flink.version}</version>
5
</dependency>

参数#

$topic Kafka 记录的 Topic 名。-
partition Kafka 记录的 partition ID。
headers 二进制 Map 类型的 Kafka 记录头（Header）
leader-epoch Kafka记录的 Leader epoch（如果可用）
offset Kafka 记录在 partition 中的 offset。
timestamp Kafka 记录的时间戳。
timestamp-type Kafka 记录的时间戳类型。可能的类型有 “NoTimestampType”， “CreateTime”（会在写入元数据时设置），或 “LogAppendTime”。
$properties.bootstrap.servers 逗号分隔的 Kafka broker 列表。
$properties.group.id kafak 组id
properties.* 可以设置和传递任意 Kafka 的配置项,后缀名必须匹配在 Kafka 配置文档中定义的配置键

一致性保证 EOS#

开启checkpoint

参数 sink.semantic

可选值：#

none 不保证任何语义
at-least-once (默认设置) 至少一次
exactly-once 精确一次

案例#

1
envTable.executeSql(
2
"""
3
        |create table add (
4
        |`UserId` bigint
5
        |) WITH (
6
        |'connector' = 'kafka',
7
        |'topic' = 'demo' ,
8
        |'format' = 'csv'
9
        |)
10
        |""".stripMargin)

format#

Csv#

pom.xml

1
<dependency>
2
  <groupId>org.apache.flink</groupId>
3
  <artifactId>flink-csv</artifactId>
4
  <version>${flink.version}</version>
5
</dependency>

参数#

csv.field-delimiter 字段分隔符 (默认’,’`)
csv.disable-quote-character 是否禁止对引用的值使用引号 (默认是 false)
csv.quote-character 用于围住字段值的引号字符 (默认")
csv.allow-comments 是否允许忽略注释行（默认不允许）—
csv.ignore-parse-errors 当解析异常时，是跳过当前字段或行，还是抛出错误失败（默认为 false，即抛出错误失败）。如果忽略字段的解析异常，则会将该字段值设置为null。
csv.array-element-delimiter 分隔数组和行元素的字符串(默认';').
csv.escape-character 转义字符(默认关闭).
csv.null-literal 是否将 “null” 字符串转化为 null 值

Json#

pom.xml

1
<dependency>
2
  <groupId>org.apache.flink</groupId>
3
  <artifactId>flink-json</artifactId>
4
  <version>${flink.version}</version>
5
</dependency>

Avro#

pom.xml

1
<dependency>
2
  <groupId>org.apache.flink</groupId>
3
  <artifactId>flink-avro</artifactId>
4
  <version>${flink.version}</version>
5
</dependency>

Flink Type#

char#

1
char
2
char(n)

n 字符串长度

varchar#

1
VARCHAR
2
VARCHAR(n)
3
STRING

n 字符串长度

BINARY#

1
BINARY
2
BINARY(n)

n 二进制字符串

VARBINARY/BYTES#

1
VARBINARY
2
VARBINARY(n)
3

4
BYTES

n 二进制字符串

DECIMAL#

1
DECIMAL
2
DECIMAL(p)
3
DECIMAL(p, s)
4

5
DEC
6
DEC(p)
7
DEC(p, s)
8

9
NUMERIC
10
NUMERIC(p)
11
NUMERIC(p, s)