Flink SQL DataGen Connector 示例

Flink SQL DataGen Connector 示例

1、概述

使用 Flink SQL DataGen Connector，可以快速地生成符合规则的测试数据，可以在不依赖真实数据的情况下进行开发和测试。

2、使用示例

创建一个名为 “users” 的表，包含 6 个字段：id、name、age、email、created_at 和 updated_at。

在表的定义中，指定各个字段的规则：

id 字段使用序列生成器，生成的范围从 1 到 1000；

name 字段的长度为 10 个字符；

age 字段的范围从 18 到 60 岁；

email 字段的长度为随机的10个字符；

created_at 和 updated_at 字段使用随机时间生成器，时间范围从 2022 年 1 月 1 日到 2022 年 12 月 31 日。

3、官网参数介绍

1）数据类型注释

Type	Supported Generators	Notes
BOOLEAN	random
CHAR	random / sequence
VARCHAR	random / sequence
STRING	random / sequence
DECIMAL	random / sequence
TINYINT	random / sequence
SMALLINT	random / sequence
INT	random / sequence
BIGINT	random / sequence
FLOAT	random / sequence
DOUBLE	random / sequence
DATE	random	Always resolves to the current date of the local machine.
TIME	random	Always resolves to the current time of the local machine.
TIMESTAMP	random	Always resolves to the current timestamp of the local machine.
TIMESTAMP_LTZ	random	Always resolves to the current timestamp of the local machine.
INTERVAL YEAR TO MONTH	random
INTERVAL DAY TO MONTH	random
ROW	random	Generates a row with random subfields.
ARRAY	random	Generates an array with random entries.
MAP	random	Generates a map with random entries.
MULTISET	random	Generates a multiset with random entries.

2）连接器参数：

参数	是否必选	默认值	数据类型	描述
connector	必须	(none)	String	指定要使用的连接器，这里是 ‘datagen’。
rows-per-second	可选	10000	Long	每秒生成的行数，用以控制数据发出速率。
fields.#.kind	可选	random	String	指定 ‘#’ 字段的生成器。可以是 ‘sequence’ 或 ‘random’。
fields.#.min	可选	(Minimum value of type)	(Type of field)	随机生成器的最小值，适用于数字类型。
fields.#.max	可选	(Maximum value of type)	(Type of field)	随机生成器的最大值，适用于数字类型。
fields.#.length	可选	100	Integer	随机生成器生成字符的长度，适用于 char、varchar、string。
fields.#.start	可选	(none)	(Type of field)	序列生成器的起始值。
fields.#.end	可选	(none)	(Type of field)	序列生成器的结束值。

4、代码示例

CREATE TABLE users (id BIGINT,name STRING,age INT,text STRING,created_at TIMESTAMP(3),updated_at TIMESTAMP(3)
) WITH ('connector' = 'datagen','fields.id.kind' = 'sequence','fields.id.start' = '1','fields.id.end' = '1000','fields.name.length' = '10','fields.age.min' = '18','fields.age.max' = '60','fields.text.length' = '5'
);

测试结果：

select * from users;