Flink：使用 Faker 和 DataGen 生成测试数据_datagen 造数

DataGen 是开源 Flink 就内置的随机数据生成器；DataGen 生成的数据仅支持随机和序列值两种，且也并不是所有的数据类型都能支持随机或序列值，例如最常见的一个需求：针对时间类型就不能生成指定区间内的单调递增的数值，相较而言，Faker 的功能要明显强于 DataGen，我们只需掌握 Faker 这一种数据生成器就足够了。

1. 安装

sudo -u flink wget https://github.com/knaufk/flink-faker/releases/download/v0.5.3/flink-faker-0.5.3.jar -P /usr/lib/flink/lib/
1

2. 示例

-- example 1: currency_rates

drop table if exists currency_rates;

create table if not exists currency_rates (
    currency_code string,
    eur_rate decimal(6,4),
    rate_time timestamp(3)
)
with (
    'connector' = 'faker',
    'fields.currency_code.expression' = '#{Currency.code}',
    'fields.eur_rate.expression' = '#{Number.randomdouble ''4'',''0'',''10''}',
    'fields.rate_time.expression' = '#{date.past ''15'',''SECONDS''}',
    'rows-per-second' = '100'
);

select * from currency_rates;

-- example 2: transactions

drop table if exists transactions;

create table if not exists transactions (
    `id` string,
    `currency_code` string,
    `total` decimal(10,2),
    `transaction_time` timestamp(3),
    watermark for `transaction_time` as transaction_time - interval '30' second
) with (
    'connector' = 'faker',
    'fields.id.expression' = '#{Internet.UUID}',
    'fields.currency_code.expression' = '#{Currency.code}',
    'fields.total.expression' = '#{Number.randomDouble ''2'',''10'',''1000''}',
    'fields.transaction_time.expression' = '#{date.past ''30'',''SECONDS''}',
    'rows-per-second' = '100'
);

select * from transactions;
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

3. 资源

Flink Faker 项目地址：https://github.com/knaufk/flink-faker/?tab=readme-ov-file

表达式文档：https://github.com/datafaker-net/datafaker