flink-sql所有表连接器-1.16_flink1.16自定义flink-connector-jdbc连接sqlserver

1. 版本说明

本文档内容基于flink-1.16.x，其他版本的整理，请查看本人博客的 flink 专栏其他文章。

2. 简介

2.1. 介绍

Flink的Table API和SQL程序可以连接到其他外部系统，用于读写批处理表和流处理表。
表source提供对存储在外部系统(如数据库、键值存储、消息队列或文件系统)中数据的访问。表sink向外部存储系统发送数据。根据source和sink的类型，它们支持不同的格式，如CSV、Avro、Parquet或ORC。

本节描述如何使用内置的连接器在Flink中注册表source和表sink。注册source或sink后，可以通过表API和SQL语句访问它。

如果你实现自己的自定义表source或sink，请查看自定义source和sink连接器页面。

2.2. 支持的连接器

Flink内置各种连接器。下表列出了所有可用的连接器。

2.3. 使用连接器

Flink支持使用SQL CREATE TABLE语句来注册表。可以定义表名、表schema和用于连接外部系统的表选项。

有关创建表的更多信息，请参阅语法部分。

下面的代码展示了如何连接到Kafka来读写JSON格式记录的完整示例。

CREATE TABLE MyUserTable (
    -- 声明表的schema
    `user` BIGINT,
    `message` STRING,
    `rowtime` TIMESTAMP(3) METADATA FROM 'timestamp',    -- 使用元数据字段来访问kafka数据的timestamp时间戳
    `proctime` AS PROCTIME(),    -- 使用计算列来定义处理时间属性
    WATERMARK FOR `rowtime` AS `rowtime` - INTERVAL '5' SECOND    -- 使用WATERMARK语句定义rowtime属性
) WITH (
    -- 定义连接的外部系统属性
    'connector' = 'kafka',
    'topic' = 'topic_name',
    'scan.startup.mode' = 'earliest-offset',
    'properties.bootstrap.servers' = 'localhost:9092',
    'format' = 'json'   -- 声明这个外部系统使用format
);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

所需的连接属性被转换为基于字符串的键值对。工厂将基于工厂标识符(本例中是kafka和json)从键值对中创建配置好的表source、表sink和相应的format格式。
在为每个组件搜索一个匹配的工厂时，会考虑所有可以通过Java的服务提供者接口(SPI)找到的工厂。

如果找不到任何工厂或找到了多个与给定属性匹配的工厂，则将抛出一个异常，并提供有关可以使用的工厂和受支持属性的附加信息。

2.4. 转换表连接器/格式资源

Flink 使用 java 的 Service Provider Interfaces (SPI) 通过他们的标识符来加载 表连接器/格式 工厂。因为 SPI 资源文件名都是 org.apache.flink.table.factories ,而且每个 表连接器/格式 的资源文件都在相同的目录 META-INF/services 下，当构建使用了多个 表连接器/格式 的项目的 uber-jar 时，这些资源文件将会相互覆盖，从而导致 Flink 加载 表连接器/格式 工厂失败。

在这种情况下，建议的方式为，通过 maven 的 shade 插件和 ServicesResourceTransformer 来转换这些存在于 META-INF/services 目录下的资源文件。下面给出同时包含连接器 flink-sql-connector-hive-3.1.2 和格式 flink-parquet 项目的 pom.xml 文件内容。

<modelVersion>4.0.0</modelVersion>
<groupId>org.example</groupId>
<artifactId>myProject</artifactId>
<version>1.0-SNAPSHOT</version>

<dependencies>
    <!--  other project dependencies  ...-->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-sql-connector-hive-3.1.2_2.12</artifactId>
        <version>1.16.0</version>
    </dependency>

    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-parquet_2.12<</artifactId>
        <version>1.16.0</version>
    </dependency>

</dependencies>

<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-shade-plugin</artifactId>
            <executions>
            <execution>
            <id>shade</id>
            <phase>package</phase>
            <goals>
                <goal>shade</goal>
            </goals>
                <configuration>
                    <transformers combine.children="append">
                    <!-- The service transformer is needed to merge META-INF/services files -->
                    <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
                    <!-- ... -->
                </transformers>
                </configuration>
            </execution>
            </executions>
        </plugin>
    </plugins>
</build>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

配置完 ServicesResourceTransformer 之后，在构建上面的项目为 uber-jar 时，META-INF/services 目录下的 表连接器/格式 的资源文件将会被合并而不是相互覆盖。

2.5. schema匹配

SQL CREATE TABLE语句的body子句定义了物理列、约束、水印的名称和类型。Flink不保存数据，因此schema定义仅声明如何将物理列从外部系统映射到Flink。
映射可能不是按名称一一对应的，这取决于格式和连接器的实现。例如，MySQL数据库表是按字段名(不区分大小写)映射的，CSV文件系统是按字段顺序映射的(字段名可以是任意的)。这些规则将根据对应的连接器来解析。

下面的示例展示了一个简单的schema，其中没有时间属性、输入/输出到表列的一对一字段映射。

CREATE TABLE MyTable (
    MyField1 INT,
    MyField2 STRING,
    MyField3 BOOLEAN
) WITH (
    ...
)
1
2
3
4
5
6
7

2.5.1. 元数据

一些连接器和格式公开了附加的元数据字段，可以在物理列后面的元数据列中访问这些字段。有关元数据列的更多信息，请参阅CREATE TABLE部分。

2.5.2. 主键

主键约束表明表中的一列或一组列是唯一的，并且它们不包含NULL值。主键唯一地标识表中的一行。

source表的主键用于优化元数据信息。sink表的主键通常用于插入更新数据。

SQL标准指定主键约束可以是ENFORCED的，也可以是NOT ENFORCED的。这将控制是否对传入/传出数据执行约束检查。
Flink本身并不拥有数据，因此唯一支持的模式是NOT ENFORCED模式。确保查询执行的主键强制约束由用户实现。

CREATE TABLE MyTable (
    MyField1 INT,
    MyField2 STRING,
    MyField3 BOOLEAN,
    PRIMARY KEY (MyField1, MyField2) NOT ENFORCED  -- 根据字段定义主键列
) WITH (
    ...
)
1
2
3
4
5
6
7
8

2.5.3. 时间属性

在处理无界流表时，时间属性是必不可少的。因此，proctime和rowtime属性都可以定义为schema的一部分。

有关Flink中的时间处理(尤其是事件时间)的更多信息，建议参阅事件时间部分。

2.5.3.1. 处理时间属性

为了在模式中声明proctime属性，可以使用计算列语法声明一个由proctime()内置函数生成的计算列。计算列是不存储在物理数据中的虚拟列。

CREATE TABLE MyTable (
    MyField1 INT,
    MyField2 STRING,
    MyField3 BOOLEAN
    MyField4 AS PROCTIME() -- 定义一个处理时间属性列
) WITH (
    ...
)
1
2
3
4
5
6
7
8

2.5.3.2. rowtime时间属性

为了控制表的事件时间行为，Flink提供了预定义的时间戳提取器和水印策略。

有关在DDL中定义时间属性的更多信息，请参阅CREATE TABLE语句。

支持以下时间戳提取器：

-- 使用已存在的TIMESTAMP(3)类型的字段作为事件时间属性
CREATE TABLE MyTable (
    ts_field TIMESTAMP(3),
    WATERMARK FOR ts_field AS ...
) WITH (
    ...
)

-- 使用系统函数、UDF、表达式来提取期望的TIMESTAMP(3)类型的事件时间属性
CREATE TABLE MyTable (
    log_ts STRING,
    ts_field AS TO_TIMESTAMP(log_ts),
    WATERMARK FOR ts_field AS ...
) WITH (
    ...
)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

支持的水印策略如下：

-- 为严格升序的事件时间属性设置水印策略。发出到目前为止观察到的最大时间戳水印。时间戳大于最大时间戳的行不属于延迟。
CREATE TABLE MyTable (
    ts_field TIMESTAMP(3),
    WATERMARK FOR ts_field AS ts_field
) WITH (
    ...
)

-- 设置升序事件时间属性的水印策略。发出到目前为止观察到的最大时间戳减去1的水印。时间戳大于或等于最大时间戳的行不属于延迟。
CREATE TABLE MyTable (
    ts_field TIMESTAMP(3),
    WATERMARK FOR ts_field AS ts_field - INTERVAL '0.001' SECOND
) WITH (
    ...
)

-- 为事件时间属性设置水印策略，这些事件时间属性在有限的时间间隔内是无序的。发出的水印是观察到的最大时间戳减去指定的延迟，例如2秒。
CREATE TABLE MyTable (
    ts_field TIMESTAMP(3),
    WATERMARK FOR ts_field AS ts_field - INTERVAL '2' SECOND
) WITH (
    ...
)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

一定要同时声明时间戳和水印。触发基于时间的操作需要水印。

2.5.4. SQL字段类型

请参阅数据类型章节，了解如何在SQL中声明类型。

3. Kafka

3.1. 介绍

支持：

• Scan Source: Unbounded
• Sink: Streaming Append Mode

Kafka 连接器提供从 Kafka topic 中消费和写入数据的能力。

3.2. 依赖

为了使用Kafka连接器，以下依赖项需要使用自动化构建工具(如Maven或SBT)的项目和SQL客户端与SQL JAR包。

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-kafka</artifactId>
  <version>1.16.0</version>
</dependency>
1
2
3
4
5

注意自己使用的 kafka 和 scala 版本。

Kafka 连接器目前并不包含在 Flink 的二进制发行版中，请查阅这里了解如何在集群运行中引用 Kafka 连接器。

3.3. 创建 Kafka 表

以下示例展示如何创建 Kafka 表：

CREATE TABLE KafkaTable (
    `user_id` BIGINT,
    `item_id` BIGINT,
    `behavior` STRING,
    `ts` TIMESTAMP(3) METADATA FROM 'timestamp'
) WITH (
    'connector' = 'kafka',
    'topic' = 'user_behavior',
    'properties.bootstrap.servers' = 'localhost:9092',
    'properties.group.id' = 'testGroup',
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'csv'
)
1
2
3
4
5
6
7
8
9
10
11
12
13

3.4. 可用的元数据

以下的连接器元数据可以在表定义中通过元数据列的形式获取。

R/W列展示元数据是可读的（R）还是可写的（W）。 只读列必须声明为VIRTUAL以在INSERT INTO操作中排除它们。

以下扩展的CREATE TABLE示例展示了使用这些元数据字段的语法：

CREATE TABLE KafkaTable (
    `event_time` TIMESTAMP(3) METADATA FROM 'timestamp',
    `partition` BIGINT METADATA VIRTUAL,
    `offset` BIGINT METADATA VIRTUAL,
    `user_id` BIGINT,
    `item_id` BIGINT,
    `behavior` STRING
) WITH (
    'connector' = 'kafka',
    'topic' = 'user_behavior',
    'properties.bootstrap.servers' = 'localhost:9092',
    'properties.group.id' = 'testGroup',
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'csv'
);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

格式元数据信息

连接器可以读取消息格式的元数据。格式元数据的配置键以**value.**作为前缀。

以下示例展示了如何获取 Kafka 和 Debezium 的元数据字段：

CREATE TABLE KafkaTable (
    `event_time` TIMESTAMP(3) METADATA FROM 'value.source.timestamp' VIRTUAL,  -- 获取Debezium格式元数据
    `origin_table` STRING METADATA FROM 'value.source.table' VIRTUAL, -- 获取Debezium格式元数据
    `partition_id` BIGINT METADATA FROM 'partition' VIRTUAL,  -- 获取kafka元数据
    `offset` BIGINT METADATA VIRTUAL,  -- 获取kafka元数据
    `user_id` BIGINT,
    `item_id` BIGINT,
    `behavior` STRING
) WITH (
    'connector' = 'kafka',
    'topic' = 'user_behavior',
    'properties.bootstrap.servers' = 'localhost:9092',
    'properties.group.id' = 'testGroup',
    'scan.startup.mode' = 'earliest-offset',
    'value.format' = 'debezium-json'
);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

3.5. 连接器参数

3.6. 特性

3.6.1. 消息键（Key）与消息体（Value）的格式

Kafka 消息的消息键和消息体部分都可以使用某种格式来序列化或反序列化成二进制数据。

3.6.1.1. 消息体格式

Kafka 的消息键是可选的，以下语句将使用消息体格式读取和写入消息，但不使用消息键格式。
format选项与value.format意义相同。所有的格式配置使用格式识别符作为前缀。

CREATE TABLE KafkaTable (
    `ts` TIMESTAMP(3) METADATA FROM 'timestamp',
    `user_id` BIGINT,
    `item_id` BIGINT,
    `behavior` STRING
) WITH (
    'connector' = 'kafka',
    ...
    'format' = 'json',
    'json.ignore-parse-errors' = 'true'
)
1
2
3
4
5
6
7
8
9
10
11

消息体格式将配置为以下的数据类型：

ROW<`user_id` BIGINT, `item_id` BIGINT, `behavior` STRING>
1

上述数据类型表示，这三个字段值将用于构建消息体内容。

3.6.1.2. 消息键和消息体格式

以下示例展示了如何同时配置和使用消息键和消息体格式。格式配置使用key或value加上格式识别符作为前缀。

CREATE TABLE KafkaTable (
    `ts` TIMESTAMP(3) METADATA FROM 'timestamp',
    `user_id` BIGINT,
    `item_id` BIGINT,
    `behavior` STRING
) WITH (
'connector' = 'kafka',
    ...
    
    'key.format' = 'json',
    'key.json.ignore-parse-errors' = 'true',
    'key.fields' = 'user_id;item_id',
    
    'value.format' = 'json',
    'value.json.fail-on-missing-field' = 'false',
    'value.fields-include' = 'ALL'
)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

消息键格式包含了在key.fields中列出的字段（使用’;'分隔）和字段顺序。 因此将配置为以下的数据类型：

ROW<`user_id` BIGINT, `item_id` BIGINT>
1

由于消息体格式配置为**‘value.fields-include’ = ‘ALL’**，所以消息键字段也会出现在消息体格式的数据类型中：

ROW<`user_id` BIGINT, `item_id` BIGINT, `behavior` STRING>
1

3.6.1.3. 重名的格式字段

如果消息键字段和消息体字段重名，则连接器无法根据表结构信息将这些列区分开。
key.fields-prefix配置项可以在表结构中为每个消息键字段名指定共同的前缀，以和消息值字段名区分开，并在配置消息键格式的时候保留原名。

以下示例展示了在消息键和消息体中同时包含version字段的情况：

CREATE TABLE KafkaTable (
    `k_version` INT,
    `k_user_id` BIGINT,
    `k_item_id` BIGINT,
    `version` INT,
    `behavior` STRING
) WITH (
    'connector' = 'kafka',
    ...
    
    'key.format' = 'json',
    'key.fields-prefix' = 'k_',
    'key.fields' = 'k_version;k_user_id;k_item_id',
    
    'value.format' = 'json',
    'value.fields-include' = 'EXCEPT_KEY'
)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

消息体格式必须配置为EXCEPT_KEY模式。格式将被配置为以下的数据类型：

消息键格式：

ROW<`version` INT, `user_id` BIGINT, `item_id` BIGINT>
1

消息体格式：

ROW<`version` INT, `behavior` STRING>
1

3.6.2. Topic 和 Partition 发现

topic和topic-pattern配置项决定了 source 消费的 topic 或 topic 的匹配规则。topic配置项可接受使用分号间隔的 topic 列表，例如topic-1;topic-2。
topic-pattern配置项使用正则表达式来探测匹配的 topic。例如topic-pattern设置为test-topic-[0-9]，则在作业启动时，所有匹配该正则表达式的 topic（以test-topic-开头，以一位数字结尾）都将被 consumer 订阅。

为允许 consumer 在作业启动之后探测到动态创建的 topic，请将scan.topic-partition-discovery.interval配置为一个非负值。这将使 consumer 能够探测匹配名称规则的 topic 和新的 partition。

请参阅Kafka DataStream 连接器文档以获取更多关于 topic 和 partition 探测的信息。

注意 topic 列表和 topic 匹配规则只适用于 source。对于 sink 端，Flink 目前只支持单一 topic。

3.6.3. 起始消费偏移量

scan.startup.mode配置项决定了 Kafka consumer 的启动模式。有效值为：

• group-offsets：从 Zookeeper/Kafka 中某个指定的消费组已提交的偏移量开始。
• earliest-offset：从可能的最早偏移量开始。
• latest-offset：从最新偏移量开始。
• timestamp：从每个 partition 指定的时间戳开始。
• specific-offsets：从每个 partition 指定的偏移量开始。

默认值group-offsets表示从 Zookeeper/Kafka 中最近一次已提交的偏移量开始消费。

如果使用了timestamp，必须使用另外一个配置项scan.startup.timestamp-millis来指定一个从格林尼治标准时间 1970 年 1 月 1 日 00:00:00.000 开始计算的毫秒单位时间戳作为起始时间。

如果使用了specific-offsets，必须使用另外一个配置项scan.startup.specific-offsets来为每个 partition 指定起始偏移量，
例如，选项值partition:0,offset:42;partition:1,offset:300表示 partition0从偏移量42开始，partition1从偏移量300开始。

3.6.4. CDC 变更日志（Changelog） Source

Flink 原生支持使用 Kafka 作为 CDC 变更日志（changelog） source。
如果 Kafka topic 中的消息是通过变更数据捕获（CDC）工具从其他数据库捕获的变更事件，则可以使用 CDC 格式将消息解析为 Flink SQL 系统中的插入（INSERT）、更新（UPDATE）、删除（DELETE）消息。

在许多情况下，变更日志（changelog） source 都是非常有用的功能，例如将数据库中的增量数据同步到其他系统，审核日志，数据库的物化视图，时态表关联数据库表的更改历史等。

Flink 提供了几种 CDC 格式：

• debezium
• canal
• maxwell

3.6.5. Sink 分区

配置项sink.partitioner指定了从 Flink 分区到 Kafka 分区的映射关系。
默认情况下，Flink 使用Kafka 默认分区器来对消息进行分区。默认分区器对没有消息键的消息使用粘性分区策略（sticky partition strategy）进行分区，对含有消息键的消息使用 murmur2 哈希算法计算分区。

为了控制数据行到分区的路由，也可以提供自定义的 sink 分区器。fixed 分区器会将同一个 Flink 分区中的消息写入同一个 Kafka 分区，从而减少网络连接的开销。

3.6.6. 一致性保证

默认情况下，如果查询在启用 checkpoint 模式下执行时，Kafka sink 按照至少一次（at-lease-once）语义保证将数据写入到 Kafka topic 中。

当 Flink checkpoint 启用时，kafka连接器可以提供精确一次（exactly-once）的语义保证。

除了启用 Flink checkpoint，还可以通过传入对应的sink.semantic选项来选择三种不同的运行模式：

• none：Flink 不保证任何语义。已经写出的记录可能会丢失或重复。
• at-least-once(默认设置)：保证没有记录会丢失（但可能会重复）。
• exactly-once：使用 Kafka 事务提供精确一次（exactly-once）语义。当使用事务向 Kafka 写入数据时，
  请将所有从 Kafka 中消费记录的应用中的消费者isolation.level配置项设置成实际所需的值（read_committed或read_uncommitted，后者为默认值）。

3.6.7. Source 按分区 Watermark

Flink 对于 Kafka 支持发送按分区的 watermark。Watermark 在 Kafka consumer 中生成。 按分区 watermark 的合并方式和在流 shuffle 时合并 Watermark 的方式一致。
Source 输出的 watermark 由读取的分区中最小的 watermark 决定。如果 topic 中的某些分区闲置，watermark 生成器将不会向前推进。
可以在表配置中设置table.exec.source.idle-timeout选项来避免上述问题。

请参阅Kafka watermark 策略以获取更多细节。

3.6.8. 安全

为了开启包括加密和认证的安全配置，你只需要在 table option （也就是 create table with 中的设置）中设置以 properties. 开头的安全配置项即可。
下面的代码片段展示如何配置使用 PLAIN 作为 SASL 机制以及提供 JAAS 配置：

CREATE TABLE KafkaTable (
    `user_id` BIGINT,
    `item_id` BIGINT,
    `behavior` STRING,
    `ts` TIMESTAMP(3) METADATA FROM 'timestamp'
) WITH (
    'connector' = 'kafka',
    ...
    'properties.security.protocol' = 'SASL_PLAINTEXT',
    'properties.sasl.mechanism' = 'PLAIN',
    'properties.sasl.jaas.config' = 'org.apache.kafka.common.security.plain.PlainLoginModule required username=\"username\" password=\"password\";'
)
1
2
3
4
5
6
7
8
9
10
11
12

下面是更复杂的案例，使用 SASL_SSL 作为安全协议，并且使用 SCRAM-SHA-256 作为 SASL 机制：

CREATE TABLE KafkaTable (
    `user_id` BIGINT,
    `item_id` BIGINT,
    `behavior` STRING,
    `ts` TIMESTAMP(3) METADATA FROM 'timestamp'
) WITH (
    'connector' = 'kafka',
    ...
    'properties.security.protocol' = 'SASL_SSL',
    -- SSL 配置
    -- 配置服务器提供的 truststore CA 的路径
    'properties.ssl.truststore.location' = '/path/to/kafka.client.truststore.jks',
    'properties.ssl.truststore.password' = 'test1234',
    -- 如果要求客户端要求身份认证，则需要配置 keystore （私钥） 路径。
    'properties.ssl.keystore.location' = '/path/to/kafka.client.keystore.jks',
    'properties.ssl.keystore.password' = 'test1234',
    -- SASL 配置
    -- 设置 SASL 机制为 SCRAM-SHA-256。
    'properties.sasl.mechanism' = 'SCRAM-SHA-256',
    -- 配置 JAAS 
    'properties.sasl.jaas.config' = 'org.apache.kafka.common.security.scram.ScramLoginModule required username=\"username\" password=\"password\";'
)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

请注意，如果你迁移了 kafka 客户端的依赖 jar 包，则登录模块 sasl.jaas.config 的 class path 可能会不一样，因此你需要重写 JAR 中该模块的 class path。
比如，如果你使用的是 SQL 客户端 JAR ，该 JAR 包已经将 kafka 客户端依赖迁移到了 org.apache.flink.kafka.shaded.org.apache.kafka ，所以普通登录模块的路径应该是 org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule 。

获取更多安全配置的扩展细节，请参考 apache kafka 文档中的安全章节。

3.7. 数据类型映射

Kafka 将消息键值以二进制进行存储，因此 Kafka 并不存在 schema 或数据类型。
Kafka 消息使用格式配置进行序列化和反序列化，例如 csv，json，avro。 因此，数据类型映射取决于使用的格式。请参阅格式页面以获取更多细节。

4. Upsert Kafka

4.1. 介绍

支持：

• Scan Source: Unbounded
• Sink: Streaming Upsert Mode

Upsert Kafka 连接器支持以 upsert 方式从 Kafka topic 中读取数据并将数据写入 Kafka topic。

作为 source，upsert-kafka 连接器生产 changelog 流，其中每条数据记录代表一个更新或删除事件。
更准确地说，如果有这个 key，则数据记录中的 value 被解释为同一 key 的最后一个 value 的 UPDATE，如果不存在相应的 key，则该更新被视为 INSERT。
用表来类比，changelog 流中的数据记录被解释为 UPSERT，也称为 INSERT/UPDATE，因为任何具有相同 key 的现有行都会被覆盖。
另外，value 为空的消息将会被视作为 DELETE 消息。

作为 sink，upsert-kafka 连接器可以消费 changelog 流。
它会将 INSERT/UPDATE_AFTER 数据作为正常的 Kafka 消息写入，并将 DELETE 数据以 value 为空的 Kafka 消息写入（表示对应 key 的消息被删除）。
Flink 将根据主键列的值对数据进行分区，从而保证主键上的消息有序，因此同一主键上的更新/删除消息将落在同一分区中。

4.2. 依赖

为了使用Upsert Kafka连接器，以下依赖项需要在使用自动化构建工具(如Maven或SBT)的项目和SQL客户端导入。

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-kafka</artifactId>
  <version>1.16.0</version>
</dependency>
1
2
3
4
5

注意自己所使用的 flink 版本。

4.3. 完整示例

下面的示例展示了如何创建和使用 Upsert Kafka 表：

CREATE TABLE pageviews_per_region (
    user_region STRING,
    pv BIGINT,
    uv BIGINT,
PRIMARY KEY (user_region) NOT ENFORCED -- 定义主键，主键字段可以有多个，这些字段共同组成key的字段
) 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,
    view_time TIMESTAMP,
    user_region STRING,
    WATERMARK FOR viewtime AS view_time - INTERVAL '2' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'page_views',
    'properties.bootstrap.servers' = '...',
    'format' = 'json'
);

-- 计算 pv、uv 并插入到 upsert-kafka sink
INSERT INTO pageviews_per_region
SELECT
    user_region,
    COUNT(*),
    COUNT(DISTINCT user_id)
FROM pageviews
GROUP BY user_region;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

注意确保在 DDL 中定义主键。

4.4. 可用的元数据

查看常规kafka连接器以获取可以用元数据信息。

4.5. 连接器参数

4.6. 特性

4.6.1. 键（key）和值（value）的格式

查看常规kafka连接器以获取键和值的格式的详细信息。注意，该连接器要求必须同时指定键和值的格式，键字段通过主键 PRIMARY KEY 约束语法派生。

下面的例子展示如何同时指定和配置键和值的格式。格式选项通过key或者value为前缀来作为格式的选项标识符。

CREATE TABLE KafkaTable (
    `ts` TIMESTAMP(3) METADATA FROM 'timestamp',
    `user_id` BIGINT,
    `item_id` BIGINT,
    `behavior` STRING,
    PRIMARY KEY (`user_id`) NOT ENFORCED
) WITH (
    'connector' = 'upsert-kafka',
    ...
    
    'key.format' = 'json',
    'key.json.ignore-parse-errors' = 'true',
    
    'value.format' = 'json',
    'value.json.fail-on-missing-field' = 'false',
    'value.fields-include' = 'EXCEPT_KEY'
)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

4.6.2. 主键约束

Upsert Kafka 始终以 upsert 方式工作，并且需要在 DDL 中定义主键。
在具有相同主键值的消息按序存储在同一个分区的前提下，在 changelog source 定义主键，意味着在物化后的 changelog 上主键具有唯一性。定义的主键将决定哪些字段出现在 Kafka 消息的 key 中。

4.6.3. 一致性保证

默认情况下，如果启用 checkpoint，Upsert Kafka sink 会保证至少一次将数据插入 Kafka topic。
这意味着，Flink 可以将具有相同 key 的重复记录写入 Kafka topic。
但由于该连接器以 upsert 的模式工作，该连接器作为 source 读入时，可以确保具有相同主键值下仅最后一条消息会生效。因此，upsert-kafka 连接器可以像 HBase sink 一样实现幂等写入。

4.6.4. 为每个分区生成相应的 watermark

Flink 支持根据 Upsert Kafka 的每个分区的数据特性发送相应的 watermark。
当使用这个特性的时候，watermark 是在 Kafka consumer 内部生成的。合并每个分区生成的 watermark 的方式和 stream shuffle 的方式是一致的。
数据源产生的 watermark 是取决于该 consumer 负责的所有分区中当前最小的 watermark。
如果该 consumer 负责的部分分区是闲置的，则整体的 watermark 并不会前进。在这种情况下，可以通过设置合适的 table.exec.source.idle-timeout 来缓解这个问题。

如想获得更多细节，请查阅 Kafka 水印策略.

4.7. 数据类型映射

Upsert Kafka 用字节存储消息的 key 和 value，因此没有 schema 或数据类型。
消息按格式进行序列化和反序列化，例如：csv、json、avro。因此数据类型映射表由指定的格式确定。请参考格式页面以获取更多详细信息。

5. Firehose

暂时不做翻译。

6. Kinesis

暂时不做翻译。

7. JDBC

7.1. 介绍

支持：

• Scan Source: Bounded
• Lookup Source: Sync Mode
• Sink: Batch Sink:
• Streaming Append & Upsert Mode

JDBC 连接器允许使用 JDBC 驱动向任意类型的关系型数据库读取或者写入数据。本文档描述了针对关系型数据库如何通过建立 JDBC 连接器来执行 SQL 查询。

如果在 DDL 中定义了主键，则JDBC sink 将以 upsert 模式与外部系统交换 UPDATE/DELETE 消息；否则，它将以 append 模式与外部系统交换消息且不支持消费 UPDATE/DELETE 消息。

7.2. 依赖

为了使用JDBC连接器，以下依赖项对于使用自动化构建工具(如Maven或SBT)的项目和带有SQL JAR包的SQL Client都是必需的。

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-jdbc</artifactId>
  <version>1.16.0</version>
</dependency>
1
2
3
4
5

注意自己使用的 flink 版本。

JDBC 连接器并不是发布版的一部分，需要自己手动添加依赖。

在连接到具体数据库时，也需要对应的驱动依赖，目前支持的驱动如下：

当前，JDBC 连接器和驱动不在 Flink 二进制发布包中，请参阅这里了解在集群上执行时何连接它们。

7.3. 创建 JDBC 表

JDBC table 可以按如下定义：

-- 在 Flink SQL 中注册一张 MySQL 表 'users'
CREATE TABLE MyUserTable (
    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" 将数据写入到 JDBC 表中
INSERT INTO MyUserTable
SELECT id, name, age, status FROM T;

-- 查看 JDBC 表中的数据
SELECT id, name, age, status FROM MyUserTable;

-- JDBC 表在时态表关联中作为维表
SELECT * FROM myTopic
LEFT JOIN MyUserTable FOR SYSTEM_TIME AS OF myTopic.proctime
ON myTopic.key = MyUserTable.id;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

7.4. 连接器参数

7.5. 过期参数

这些过期的参数已经被上面列出的新参数代替，并且将会在未来被删除。请考虑优先使用新参数。

7.6. 特性

7.6.1. 键处理

当写入数据到外部数据库时，Flink 会使用 DDL 中定义的主键。如果定义了主键，则连接器将以 upsert 模式工作，否则连接器将以 append 模式工作。

在 upsert 模式下，Flink 将根据主键判断是插入新行还是更新已存在的行，这种方式可以确保幂等性。
为了确保输出结果是符合预期的，推荐为表定义主键并且确保主键是底层数据库中表的唯一键或主键。
在 append 模式下，Flink 会把所有记录解释为 INSERT 消息，如果违反了底层数据库中主键或者唯一约束，INSERT 插入可能会失败。

有关 PRIMARY KEY 语法的更多详细信息，请参见 CREATE TABLE DDL。

7.6.2. 分区扫描

为了在并行 Source task 实例中加速读取数据，Flink 为 JDBC table 提供了分区扫描的特性。

如果下述分区扫描参数中的任一项被指定，则下述所有的分区扫描参数必须都被指定。
这些参数描述了在多个 task 并行读取数据时如何对表进行分区。 scan.partition.column 必须是相关表中的数字、日期或时间戳列。
注意，scan.partition.lower-bound 和 scan.partition.upper-bound 用于决定分区的起始位置和结束位置，以过滤表中的数据。
如果是批处理作业，也可以在提交 flink 作业之前获取最大值和最小值。

• scan.partition.column：用于进行分区的列名。
• scan.partition.num：分区数。
• scan.partition.lower-bound：第一个分区的最小值。
• scan.partition.upper-bound：最后一个分区的最大值。

注意

最大值和最小值设置，对于数据库中的时间和日期字段，该值应该使用毫秒值，而不是字符串。

flink会根据分区数对整个分区范围进行切分，然后将切分后的所有区间分配给所有source并行度。
flink生成的查询sql语句最后面会添加 where 分区字段过滤条件，使用 between 关键字。

7.6.3. Lookup Cache

JDBC 连接器可以在时态表关联中作为一个可 lookup 的 source (又称为维表)，当前只支持同步的查找模式。

默认情况下，lookup cache 是未启用的，可以通过设置 lookup.cache.max-rows 和 lookup.cache.ttl 参数来启用。

lookup cache 的主要目的是用于提高时态表关联 JDBC 连接器的性能。
默认情况下，lookup cache 不开启，所以所有请求都会发送到外部数据库。当 lookup cache 被启用时，每个进程（即 TaskManager）将维护一个缓存。
Flink 将优先查找缓存，只有当缓存未查找到时才向外部数据库发送请求，并使用返回的数据更新缓存。
当缓存命中最大缓存行 lookup.cache.max-rows 或当行超过最大存活时间 lookup.cache.ttl 时，缓存中最老的行将被设置为已过期。
缓存中的记录可能不是最新的，用户可以将 lookup.cache.ttl 设置为一个更小的值以获得更新的刷新数据，但这可能会增加发送到数据库的请求数。所以要做好吞吐量和正确性之间的平衡。

默认情况下，flink 会缓存主键查询为空的结果，你可以设置参数 lookup.cache.caching-missing-key 为 false 来改变这个行为。

7.6.4. 幂等写入

如果在 DDL 中定义了主键，JDBC sink 将使用 upsert 语义而不是普通的 INSERT 语义。upsert 语义指的是如果数据主键值在数据库中已存在，则更新现有行；如果不存在，则插入新行，这种方式确保了幂等性。

如果出现故障，Flink 作业会从上次成功的 checkpoint 恢复并重新处理，但这可能导致在恢复过程中重复处理消息。强烈推荐使用 upsert 模式，因为如果需要重复处理记录，它有助于避免违反数据库主键约束和产生重复数据。

除了故障恢复场景外，数据源（kafka topic）也可能随着时间的推移自然地包含多个具有相同主键的记录，这使得 upsert 模式是符合用户期待的。

upsert 没有标准的语法，下表描述了不同数据库的 DML 语法：

7.6.5. JDBC Catalog

JdbcCatalog 允许用户使用 flink 通过 JDBC 协议去连接关系型数据库。

目前已经有两个 JDBC catalog 的实现，Postgres Catalog 和 MySQL Catalog 。他们支持下面的 catalog 函数，其他函数目前还不支持。

// Postgres 和 MySQL Catalog 支持的函数
databaseExists(String databaseName);
listDatabases();
getDatabase(String databaseName);
listTables(String databaseName);
getTable(ObjectPath tablePath);
tableExists(ObjectPath tablePath);
1
2
3
4
5
6
7

其他的 catalog 函数目前还不支持。

7.6.5.1. 使用 JDBC catalog

该章节描述怎么创建并使用 Postgres Catalog 或 MySQL Catalog 。怎么添加 JDBC 连接器和相关的驱动，请参考上面的依赖章节。

JDBC catalog 支持下面的选项配置：

• name：必选，catalog的名称。
• default-database：必选，连快将诶的默认数据库。
• username：必选，Postgres/MySQL 账户的用户名。
• password：必选，账户的密码。
• base-url：必选，不要包含数据库名称。
  • 对于 Postgres Catalog ，该参数应该为：jdbc:postgresql://<ip>:<port>
  • 对于 MySQL Catalog ，该参数应该为：jdbc:mysql://<ip>:<port>

SQL：

CREATE CATALOG my_catalog WITH(
    'type' = 'jdbc',
    'default-database' = '...',
    'username' = '...',
    'password' = '...',
    'base-url' = '...'
);

USE CATALOG my_catalog;
1
2
3
4
5
6
7
8
9

java

EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
TableEnvironment tableEnv = TableEnvironment.create(settings);

String name            = "my_catalog";
String defaultDatabase = "mydb";
String username        = "...";
String password        = "...";
String baseUrl         = "..."

JdbcCatalog catalog = new JdbcCatalog(name, defaultDatabase, username, password, baseUrl);
tableEnv.registerCatalog("my_catalog", catalog);

// 设置 JdbcCatalog 为会话的当前 catalog
tableEnv.useCatalog("my_catalog");
1
2
3
4
5
6
7
8
9
10
11
12
13
14

scala

val settings = EnvironmentSettings.inStreamingMode()
val tableEnv = TableEnvironment.create(settings)

val name            = "my_catalog"
val defaultDatabase = "mydb"
val username        = "..."
val password        = "..."
val baseUrl         = "..."

val catalog = new JdbcCatalog(name, defaultDatabase, username, password, baseUrl)
tableEnv.registerCatalog("my_catalog", catalog)

// 设置 JdbcCatalog 为会话的当前 catalog
tableEnv.useCatalog("my_catalog")
1
2
3
4
5
6
7
8
9
10
11
12
13
14

python

from pyflink.table.catalog import JdbcCatalog

environment_settings = EnvironmentSettings.in_streaming_mode()
t_env = TableEnvironment.create(environment_settings)

name = "my_catalog"
default_database = "mydb"
username = "..."
password = "..."
base_url = "..."

catalog = JdbcCatalog(name, default_database, username, password, base_url)
t_env.register_catalog("my_catalog", catalog)

# 设置 JdbcCatalog 为会话的当前 catalog
t_env.use_catalog("my_catalog")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

yaml

execution:
    ...
    current-catalog: my_catalog  # 设置 JdbcCatalog 为会话的当前 catalog
    current-database: mydb

catalogs:
   - name: my_catalog
     type: jdbc
     default-database: mydb
     username: ...
     password: ...
     base-url: ...
1
2
3
4
5
6
7
8
9
10
11
12

7.6.5.2. JDBC Catalog for PostgreSQL

PostgreSQL 元空间映射

PostgreSQL 基于数据库有一个额外的命名空间作为 schema 。一个 Postgres 示例可以有多个数据库，每个数据可以有多个 schema ，默认的 schema 名为 public ，每个 schema 可以有多张表。
在 flink 中，当查询注册到 Postgres catalog 中的表时，用户可以使用 schema_name.table_name 或者是只使用 table_name 。schema_name 是可选的，默认为 public 。

在 flink catalog 和 Postgres 之间的元空间映射如下：

Flink 中的 Postgres 表的完整路径应该是：

<catalog>.<db>.`<schema.table>`
1

如果指定了 schema，请注意需要转义<schema.table>，也就是使用转义字符 ` 将其括起来。

下面是一些访问 Postgres 表的案例：

-- 浏览名为 'public' 的 schema 下的 'test_table' 表，使用默认的 schema，schema 名称可以被省略。
SELECT * FROM mypg.mydb.test_table;
SELECT * FROM mydb.test_table;
SELECT * FROM test_table;

-- 浏览名为 'custom_schema' 的 schema 下的 'test_table2' 表，自定义 schema 不能被省略，而且必须和表一起被转义。
SELECT * FROM mypg.mydb.`custom_schema.test_table2`
SELECT * FROM mydb.`custom_schema.test_table2`;
SELECT * FROM `custom_schema.test_table2`;
1
2
3
4
5
6
7
8
9

7.6.5.3. JDBC Catalog for MySQL

MySQL 元空间映射

MySQL 实例中的数据库和注册到 MySQL catalog 中的数据库有相同的映射级别。一个 MySQL 实例可以有多个数据库，每个数据库可以有多张表。
在 flink 中，当查询注册到 MySQL catalog 中的表时，用户可以使用 database.table_name 或者只指定 table_name 。默认的数据库名为创建 MySQL Catalog 时指定的默认数据库。

flink Catalog 和 MySQL Catalog 之间的元空间映射关系如下：

flink 中的 MySQL 表的完全路径应该为：

`<catalog>`.`<db>`.`<table>`
1

下面是一些访问 MySQL 表的案例：

-- 浏览 'test_table' 表，默认数据库为 'mydb'
SELECT * FROM mysql_catalog.mydb.test_table;
SELECT * FROM mydb.test_table;
SELECT * FROM test_table;

-- 浏览指定数据库下的 'test_table' 表。
SELECT * FROM mysql_catalog.given_database.test_table2;
SELECT * FROM given_database.test_table2;
1
2
3
4
5
6
7
8

7.7. 数据类型映射

Flink 支持连接到多个使用方言（dialect）的数据库，如 MySQL、PostgresSQL、Derby 等。
其中，Derby 通常是用于测试目的。下表列出了从关系数据库数据类型到 Flink SQL 数据类型的类型映射，映射表可以使得在 Flink 中定义 JDBC 表更加简单。

8. Elasticsearch

8.1. 介绍

支持：

• Sink: Batch
• Sink: Streaming Append & Upsert Mode

Elasticsearch 连接器允许将数据写入到 Elasticsearch 引擎的索引中。

连接器可以工作在 upsert 模式下，使用 DDL 中定义的主键与外部系统交换 UPDATE/DELETE 消息。

如果 DDL 中没有定义主键，则连接器只能工作在 append 模式，只能与外部系统交换 INSERT 消息。

8.2. 依赖

为了使用Elasticsearch连接器，以下依赖项对于使用自动化构建工具(如Maven或SBT)的项目和带有SQL JAR包的SQL Client都是必需的。

6.x

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-elasticsearch6</artifactId>
  <version>1.16.0</version>
</dependency>
1
2
3
4
5

7.x and later versions

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-elasticsearch7</artifactId>
  <version>1.16.0</version>
</dependency>
1
2
3
4
5

注意自己使用的 flink 和 scala 版本。

8.3. 创建 Elasticsearch 表

以下示例展示如何创建 Elasticsearch sink 表：

CREATE TABLE myUserTable (
    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'
);
1
2
3
4
5
6
7
8
9
10
11

8.4. 连接器参数

8.5. 特性

8.5.1. Key 处理

Elasticsearch sink 可以根据是否定义了主键来确定是在 upsert 模式还是 append 模式下工作。
如果定义了主键，Elasticsearch sink 将以 upsert 模式工作，该模式可以消费包含 UPDATE/DELETE 的消息。 如果未定义主键，Elasticsearch sink 将以 append 模式工作，该模式只能消费包含 INSERT 的消息。

在 Elasticsearch 连接器中，主键用于计算 Elasticsearch 的文档 id，文档 id 为最多 512 字节且不包含空格的字符串。
Elasticsearch 连接器通过使用 document-id.key-delimiter 指定的键分隔符按照 DDL 中定义的顺序连接所有主键字段，为每一行记录生成一个文档 ID 字符串。
某些类型不允许作为主键字段，因为它们没有对应的字符串表示形式，例如，BYTES，ROW，ARRAY，MAP 等。 如果未指定主键，Elasticsearch 将自动生成文档 id。

有关 PRIMARY KEY 语法的更多详细信息，请参见 CREATE TABLE DDL。

8.5.2. 动态索引

Elasticsearch sink 同时支持静态索引和动态索引。

如果想使用静态索引，则 index 选项值应为纯字符串，例如 myusers，所有记录都将被写入到 myusers 索引中。

如果想使用动态索引，你可以使用 {field_name} 来引用记录中的字段值来动态生成目标索引。
你也可以使用 {field_name|date_format_string} 将 TIMESTAMP/DATE/TIME 类型的字段值转换为 date_format_string 指定的格式。

date_format_string 与 Java 的 DateTimeFormatter 兼容。
例如，如果选项值设置为 myusers-{log_ts|yyyy-MM-dd}，则 log_ts 字段值为 2020-03-27 12:25:55 的记录将被写入到 myusers-2020-03-27 索引中。

你也可以使用 {now()|date_format_string} 来转换当前时间为指定的 date_format_string 格式化字符串。对应的 now() 时间类型为 TIMESTAMP_WITH_LTZ 。当格式化系统时间为字符串时，会使用通过 table.local-time-zone 指定的时区。你可以使用：NOW()、now()、CURRENT_TIMESTAMP、current_timestamp。

注：当使用当前系统时间来生成动态索引时，对于 changelog 流，无法保证有相同唯一主键的数据可以生成相同的索引名称。因此，基于系统时间生成动态索引只支持 append only 流。

8.6. 数据类型映射

Elasticsearch 将文档存储在 JSON 字符串中。因此数据类型映射介于 Flink 数据类型和 JSON 数据类型之间。
Flink 为 Elasticsearch 连接器使用内置的 json 格式。更多类型映射的详细信息，请参阅 JSON Format 页面。

9. FileSystem

9.1. 介绍

该连接器通过Flink FileSystem abstraction提供对文件系统中分区文件的访问。

文件系统连接器被包含在flink中，不需要添加额外的依赖即可使用。从文件系统中读写行数据，需要相对应的定义格式。

文件系统连接器允许从本地或分布式文件系统读取或写入数据，下面是一个文件系统表的定义：

CREATE TABLE MyUserTable (
  column_name1 INT,
  column_name2 STRING,
  ...
  part_name1 INT,
  part_name2 STRING
)
PARTITIONED BY (part_name1, part_name2)
WITH (
  'connector' = 'filesystem',                   -- 必填: 指定连接器名称
  'path' = 'file:///path/to/whatever',          -- 必填: 目录路径
  'format' = '...',                             -- 必填: 文件系统连接器要求指定一个format格式化
  'partition.default-name' = '...',             -- 可选: 如果动态分区字段值为null/空字符串，则使用指定的默认分区名称
  'sink.shuffle-by-partition.enable' = '...',   -- 可选：在sink阶段开启对动态分区文件数据的shuffle，开启之后可以减少写出文件的数量，但是有可能造成数据倾斜。默认为false。
  ...
);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

请确保包含了flink文件系统需要的依赖。

该文件系统连接器和以前传统的文件系统连接器有很多不同：path参数指定的是目录而不是一个文件，而且你无法获取指定路径中你声明的一个可读文件。

9.2. 分区文件

文件系统分区支持使用标准的hive format格式，而且，它不要求分区被预注册在表的catalog中。分区通过目录结构来进行发现和推断。比如，下面基于目录的表分区将会被推断为包含日期和小时分区。

path
└── datetime=2019-08-25
    └── hour=11
        ├── part-0.parquet
        ├── part-1.parquet
    └── hour=12
        ├── part-0.parquet
└── datetime=2019-08-26
    └── hour=6
        ├── part-0.parquet
1
2
3
4
5
6
7
8
9
10

文件系统表同时支持插入和覆盖。查看INSERT Statement章节。当使用insert overwrite覆盖一个分区表时，只有相关联的分区被覆盖，而不是整张表。

9.3. 文件format

文件系统连接器支持多种format格式：

• CSV: RFC-4180. 未压缩
• JSON: 注意，文件系统的JSON格式并不是标准的JSON文件，而是未压缩的newline delimited JSON。
• Avro: Apache Avro. 支持通过配置avro.codec来支持压缩。
• Parquet: Apache Parquet. 兼容Hive.
• Orc: Apache Orc. 兼容Hive.
• Debezium-JSON: debezium-json.
• Canal-JSON: canal-json.
• Raw: raw.

9.4. Source

file system 连接器在单个表中可以被用于读取单个文件，或者是整个目录。

当使用目录作为 source 路径时，目录中的文件并没有定义好的读取顺序。

9.4.1. 目录监控

默认情况下，file system 连接器是有界的，该连接器只会读取一次配置的目录，然后关闭它。

你可以通过配置 option source.monitor-interval 选项配置持续的目录监控：

9.4.2. 可用元数据

下面的连接器元数据可以通过被定义为表的元数据字段来访问，所有的元数据都是只读的。

下面的代码片段展示了 CREATE TABLE 案例如何访问元数据属性：

CREATE TABLE MyUserTableWithFilepath (
    column_name1 INT,
    column_name2 STRING,
    `file.path` STRING NOT NULL METADATA
) WITH (
    'connector' = 'filesystem',
    'path' = 'file:///path/to/whatever',
    'format' = 'json'
)
1
2
3
4
5
6
7
8
9

9.5. Streaming Sink

文件系统连接器基于Streaming File Sink 写入记录到文件以支持文件系统连接器流式写入。行编码格式支持csv和json。块编码格式支持parquet、orc和avro。

可以通过sql直接写入，插入流数据到不分区的表中。如果是分区表，可以配置分区关联操作。具体查看下面的分区提交章节。

9.5.1. 滚动策略

数据通过分区目录会被切分为多个文件。每个分区将包含其对应sink子任务接收到数据之后写入的至少一个文件，正在处理的文件将会根据配置的滚动策略来关闭并成为分区中的一个文件。
文件的滚动策略基于大小、文件可以被打开的最大超时时间间隔来配置。

注：对于块格式（parquet、orc、avro），滚动策略将会根据checkpoint间隔来控制大小和他们的数量，checkpoint决定文件的写入完成。

注：对于行格式（csv、json），如果想查看文件是否在文件系统中存在，并且不想等待过长的时间，
则可以在连接器配置 sink.rolling-policy.file-size 和 sink.rolling-policy.rollover-interval ，
并且在flink-conf.yaml中设置 execution.checkpointing.interval 参数。

对于其他的格式（avro、orc），可以只在flink-conf.yaml中配置execution.checkpointing.interval参数。

9.5.2. 文件压缩

文件系统sink支持文件压缩，该特性允许应用程序设置更小的checkpoint间隔，而不会产生很多的文件。

如果开启，文件压缩将会基于目标文件大小合并多个小文件为大文件。在生产生运行文件压缩时，需要注意以下问题：

• 只有单个checkpoint中的文件可以被合并，因此，至少有和checkpoint次数相同的文件被生成。
• 文件在被合并之前是不可见的，因此文件可见时间为：checkpoint间隔+压缩时间。
• 如果压缩运行时间过长，则将会造成任务的反压，并且增加checkpoint的时间。

9.5.3. 分区提交

通常来说，写入分区之后通知下游应用程序是非常必要的。比如：增加分区信息到hive的元数据，或者是在分区目录中写入一个 _SUCCESS 文件。
文件系统sink连接器提供了分区提交特性，以允许配置自定义策略。提交行为基于合并的触发器和策略。

Trigger触发器：分区提交的时间可以通过水印或处理时间来确定。

Policy策略：如何提交一个分区，内奸策略支持通过success文件和元数据提交，也可以自定义实现策略。比如触发hive的指标分区，或者是和并小文件等等。

注：分区提交只在动态分区插入时起作用。

9.5.3.1. 分区提交触发器

定义何时提交分区，提供分区提交触发器：

有两种触发器类型：

• 第一个是分区的处理时间，既不要求分区时间提取，也不要求水印生成。该触发器根据分区的创建时间和当前系统时间触发分区提交。该触发器更常用，但不是很精确。比如，数据延迟或失败，将会导致不成熟的分区提交。
• 第二个是根据水印和从分区中提取的时间来触发分区提交。该触发器要求任务有水印生成，并且分区根据时间来划分，比如按小时或按天分区。

如果想要下游尽快看到新分区，而不管数据写入是否完成：

• ‘sink.partition-commit.trigger’=‘process-time’ （默认值）
• ‘sink.partition-commit.delay’=‘0s’ （默认值），分区一旦写入数据，将会立即提交。注：分区可能会被提交多次。

如果想要下游在数据写入完成之后看到分区，并且job任务有水印生成，则可以通过分区值来提取时间：

• ‘sink.partition-commit.trigger’=‘partition-time’
• ‘sink.partition-commit.delay’=‘1h’ （如果分区为小时分区，则使用 1h，取决于分区时间类型）这是提交分区更准确的方式。它将尝试在数据写入完成之后再提交分区。

如果想要下游在数据写入完成之后看到分区，但是没有水印，或者是无法从分区值提取时间：

• ‘ink.partition-commit.trigger’=‘process-time’ （默认值）
• ‘sink.partition-commit.delay’=‘1h’ （如果分区为小时分区，则使用 1h，取决于分区时间类型）尝试准确的提交分区，但是迟到的数据或者是失败将会导致不成熟的分区提交。

迟到数据处理：支持写入分区的记录将会被写入已经提交的分区，并且该分区提交将会被再次触发。

9.5.3.2. 分区时间提取

时间提取器定义定分区值提取时间的方式。

默认提取器基于分区属性和时间戳默认组成。也可以通过实现 PartitionTimeExtractor 接口来完全自定义分区提取器。

public class HourPartTimeExtractor implements PartitionTimeExtractor {
    @Override
    public LocalDateTime extract(List<String> keys, List<String> values) {
        String dt = values.get(0);
        String hour = values.get(1);
        return Timestamp.valueOf(dt + " " + hour + ":00:00").toLocalDateTime();
    }
}
1
2
3
4
5
6
7
8

9.5.3.3. 分区提交策略

分区提交策略定义分区提交时执行哪些操作：

• 第一个是元数据，只有hive表支持元数据策略，文件系统通过目录结构管理分区。
• 第二个是success文件，在分区对一个的目录下写一个空文件。

也可以向下面一样扩展提交策略实现：

public class AnalysisCommitPolicy implements PartitionCommitPolicy {
    private HiveShell hiveShell;

    @Override
    public void commit(Context context) throws Exception {
        if (hiveShell == null) {
            hiveShell = createHiveShell(context.catalogName());
        }

        hiveShell.execute(String.format(
                "ALTER TABLE %s ADD IF NOT EXISTS PARTITION (%s = '%s') location '%s'",
                context.tableName(),
                context.partitionKeys().get(0),
                context.partitionValues().get(0),
                context.partitionPath()));
        hiveShell.execute(String.format(
                "ANALYZE TABLE %s PARTITION (%s = '%s') COMPUTE STATISTICS FOR COLUMNS",
                context.tableName(),
                context.partitionKeys().get(0),
                context.partitionValues().get(0)));
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

9.6. sink并行度

写入文件到外部文件系统的并行度（包括hive），可以通过表的option选项来配置，流模式和批模式都支持这么做。
默认情况下，slink的并行度和上游链在一起的算子并行度一致。如果配置了和上游算子不同的并行度，则写入文件算子的并行度将使用配置的并行度。

注：目前，配置sink并行度只支持上游算子为仅插入INERT-ONLY类型的变更日志模式，否则将抛出异常。

9.7. 完整案例

下面的例子展示文件系统连接器如何通过流查询从kafka读取数据，然后写入文件系统，并且通过批查询从文件系统中读取写入的数据。

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 (...);

CREATE TABLE fs_table (
  user_id STRING,
  order_amount DOUBLE,
  dt STRING,
  `hour` STRING
) PARTITIONED BY (dt, `hour`) WITH (
  'connector'='filesystem',
  'path'='...',
  'format'='parquet',
  'sink.partition-commit.delay'='1 h',
  'sink.partition-commit.policy.kind'='success-file'
);

-- streaming sql, insert into file system table
INSERT INTO fs_table 
SELECT 
    user_id, 
    order_amount, 
    DATE_FORMAT(log_ts, 'yyyy-MM-dd'),
    DATE_FORMAT(log_ts, 'HH') 
FROM kafka_table;

-- 批式sql，查询指定分区下的数据
SELECT * FROM fs_table WHERE dt='2020-05-20' and `hour`='12';
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

如果水印定义在TIMESTAMP_LTZ类型的字段上，并且被用于分区提交时间，则sink.partition-commit.watermark-time-zone配置必须设置为会话时间分区，否则分区提交将会晚几个小时。

CREATE TABLE kafka_table (
  user_id STRING,
  order_amount DOUBLE,
  ts BIGINT, -- 毫秒值
  ts_ltz AS TO_TIMESTAMP_LTZ(ts, 3),
  WATERMARK FOR ts_ltz AS ts_ltz - INTERVAL '5' SECOND -- 在TIMESTAMP_LTZ字段上定义水印
) WITH (...);

CREATE TABLE fs_table (
  user_id STRING,
  order_amount DOUBLE,
  dt STRING,
  `hour` STRING
) PARTITIONED BY (dt, `hour`) WITH (
  'connector'='filesystem',
  'path'='...',
  'format'='parquet',
  'partition.time-extractor.timestamp-pattern'='$dt $hour:00:00',
  'sink.partition-commit.delay'='1 h',
  'sink.partition-commit.trigger'='partition-time',
  'sink.partition-commit.watermark-time-zone'='Asia/Shanghai', -- 表名用户配置的时区为：'Asia/Shanghai'
  'sink.partition-commit.policy.kind'='success-file'
);

-- 流式sql，插入数据到文件系统
INSERT INTO fs_table 
SELECT 
    user_id, 
    order_amount, 
    DATE_FORMAT(ts_ltz, 'yyyy-MM-dd'),
    DATE_FORMAT(ts_ltz, 'HH') 
FROM kafka_table;

-- 批式sql，查询指定分区下的数据
SELECT * FROM fs_table WHERE dt='2020-05-20' and `hour`='12';
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

10. Hbase

10.1. 介绍

支持：

• Scan Source: Bounded
• Lookup Source: Sync Mode
• Sink: Batch
• Sink: Streaming Upsert Mode

HBase 连接器支持读取和写入 HBase 集群。本文档介绍如何使用 HBase 连接器基于 HBase 进行 SQL 查询。

HBase 连接器在 upsert 模式下运行，可以使用 DDL 中定义的主键与外部系统交换更新操作消息。但是主键只能基于 HBase 的 rowkey 字段定义。如果没有声明主键，HBase 连接器默认取 rowkey 作为主键。

10.2. 依赖

为了使用HBase连接器，要求下面的依赖添加到通过自动构建工具（比如maven和SBT）构建的项目中，或者是SQL的客户端。

1.4.x

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-hbase-1.4</artifactId>
  <version>1.16.0</version>
</dependency>
1
2
3
4
5

2.2.x

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-hbase-2.2</artifactId>
  <version>1.16.0</version>
</dependency>
1
2
3
4
5

注意自己使用的 flink 和 scala 版本。

10.3. 如何使用 HBase 表

所有 HBase 表的列簇必须定义为 ROW 类型，字段名对应列簇名（column family），嵌套的字段名对应列限定符名（column qualifier）。
用户只需在表结构中声明查询中使用的列簇和列限定符。
除了 ROW 类型的列，剩下的原子数据类型字段（比如，STRING, BIGINT）将被识别为 HBase 的 rowkey，一张表中只能声明一个 rowkey。rowkey 字段的名字可以是任意的，如果是保留关键字，需要用反引号。

-- 在 Flink SQL 中注册 HBase 表 "mytable"
CREATE TABLE hTable (
 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'
);

-- 用 ROW(...) 构造函数构造列簇，并往 HBase 表写数据。
-- 假设 "T" 的表结构是 [rowkey, f1q1, f2q2, f2q3, f3q4, f3q5, f3q6]
INSERT INTO hTable
SELECT rowkey, ROW(f1q1), ROW(f2q2, f2q3), ROW(f3q4, f3q5, f3q6) FROM T;

-- 从 HBase 表扫描数据
SELECT rowkey, family1, family3.q4, family3.q6 FROM hTable;

-- temporal join HBase 表，将 HBase 表作为维表
SELECT * FROM myTopic
LEFT JOIN hTable FOR SYSTEM_TIME AS OF myTopic.proctime
ON myTopic.key = hTable.rowkey;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

10.4. 连接器参数

10.5. 过期参数

这些过期的参数已经被上面列出的新参数代替了，并且在将来会被移除，请优先考虑使用新的参数。

10.6. 数据类型映射表

HBase 以字节数组存储所有数据。在读和写过程中要序列化和反序列化数据。

Flink 的 HBase 连接器利用 HBase（Hadoop) 的工具类 org.apache.hadoop.hbase.util.Bytes 进行字节数组和 Flink 数据类型转换。

Flink 的 HBase 连接器将所有数据类型（除字符串外）的null 值编码成空字节。对于字符串类型，null 值的字面值由null-string-literal选项值决定。

数据类型映射表如下：

11. DataGen

11.1. 介绍

支持：

• Scan Source: Bounded
• Scan Source: UnBounded

DataGen连接器允许通基内存的数据生成来创建表，这对于本地查询，而不是外部系统查询来说是非常有用的，比如kafka。表可以包含Computed Column syntax字段计算语法，以支持更灵活的数据生成。

DataGen是内建连接器，无需添加额外依赖。

11.2. 使用

默认情况下，DataGen表将创建无限数量的数据行，并且每个字段都是随机值。对于可变大小的类型，比如char/varchar/string/array/map/multiset，可以指定他们的长度。另外也可以通过指定总行数，来创建一个有界表。

flink也提供了序列化生成器，用户可以指定序列的开始和结束之。如果表的某个字段为序列类型，则表将会成为有界表，当第一个字段值生成到他对应的结束值时数据生成结束。

时间类型通常为本地机器的当前系统时间。

CREATE TABLE Orders (
    order_number BIGINT,
    price        DECIMAL(32,2),
    buyer        ROW<first_name STRING, last_name STRING>,
    order_time   TIMESTAMP(3)
) WITH (
  'connector' = 'datagen'
)
1
2
3
4
5
6
7
8

通常情况下，数据生成连接器和LINK子句一起使用来模拟物理表。

CREATE TABLE Orders (
    order_number BIGINT,
    price        DECIMAL(32,2),
    buyer        ROW<first_name STRING, last_name STRING>,
    order_time   TIMESTAMP(3)
) WITH (...)

-- 创建一个模拟表
CREATE TEMPORARY TABLE GenOrders
WITH (
    'connector' = 'datagen',
    'number-of-rows' = '10'
)
LIKE Orders (EXCLUDING ALL)
1
2
3
4
5
6
7
8
9
10
11
12
13
14

11.3. 数据类型

11.4. 连接器选项

注：上面选项中的 # ，在实际使用时，要替换为创建的表的某个字段名称，表示对这个字段设置属性。

12. Print

12.1. 介绍

支持：

• Sink

Print 连接器允许将每一行写入标准输出流或者标准错误流。

设计目的：

• 简单的流作业测试。
• 对生产调试带来极大便利。

四种 format 选项：

输出字符串格式为 $row_kind(f0,f1,f2…)，row_kind是一个 RowKind 类型的短字符串，例如：+I(1,1) 。

Print 连接器是内置的。

注意：在任务运行时使用 Print Sinks 打印记录，需要注意观察任务日志。

12.2. 创建Print表

CREATE TABLE print_table (
    f0 INT,
    f1 INT,
    f2 STRING,
    f3 DOUBLE
) WITH (
    'connector' = 'print'
)
1
2
3
4
5
6
7
8

也可以通过 LIKE 子句 基于已有表的结构去创建新表。

CREATE TABLE print_table WITH ('connector' = 'print')
LIKE source_table (EXCLUDING ALL)
1
2

12.3. 连接器参数

13. BlackHole

13.1. 介绍

支持：

• Sink: Bounded
• Sink: UnBounded

BlackHole 连接器允许接收所有输入记录。它被设计用于：

• 高性能测试。
• UDF 输出，而不是实质性 sink。

就像类 Unix 操作系统上的 /dev/null。

BlackHole 连接器是内置的。

13.2. 创建 BlackHole 表

CREATE TABLE blackhole_table (
    f0 INT,
    f1 INT,
    f2 STRING,
    f3 DOUBLE
) WITH (
    'connector' = 'blackhole'
);
1
2
3
4
5
6
7
8

也可以基于现有模式使用 LIKE 子句 创建。

CREATE TABLE blackhole_table WITH ('connector' = 'blackhole')
LIKE source_table (EXCLUDING ALL)
1
2

13.3. 连接器选项