This topic provides the DDL syntax that is used to create a Datagen source table, describes the parameters in the WITH clause, and provides sample code.
What is a Datagen source table?
A Datagen source table is a built-in connector of the Flink system. This connector periodically generates random data of the type that corresponds to the Datagen source table. If an error is returned when you create a source table of another type, but you cannot determine whether it is caused by a system error or an invalid setting of a parameter in the WITH clause of the source table, you can change the value of connector to datagen and click Run. If no error is returned, Realtime Compute for Apache Flink is normal. You must check the settings of parameters in the WITH clause.
Limits
Only Realtime Compute for Apache Flink that uses Ververica Runtime (VVR) 2.0.0 or later supports the Datagen connector.
DDL syntax
CREATE TABLE datagen_source (
name VARCHAR,
score BIGINT
) WITH (
'connector' = 'datagen'
);
Parameters in the WITH clause
Parameter | Description | Required | Remarks |
---|---|---|---|
connector | The type of the source table. | Yes | Set the value to datagen. |
rows-per-second | The rate at which random data is generated. | No | Default value: 10000. Unit: records per second. |
fields.#.length | The length of the generated random string. | No | Default value: 100. The CHAR, VARCHAR, and STRING data types are supported. |
fields.#.kind |
|
No | N/A. |
fields.#.max | The maximum value of the random number generator. | No | Only numeric values are supported. |
fields.#.min | The minimum value of the random number generator. | No | Only numeric values are supported. |
fields.#.start | The start value of the sequence generator. | No | N/A. |
fields.#.end | The end value of the sequence generator. | No | N/A. |
Sample code
CREATE TEMPORARY table datagen_source(
name VARCHAR
) WITH (
'connector' = 'datagen'
);
CREATE TEMPORARY table datahub_sink(
name VARCHAR
) WITH (
'connector'='datahub',
'endpoint'='<yourEndpoint>',
'project'='<yourProject>',
'topic'='<yourTopic>',
'accessId'='<yourAccessId>',
'accessKey'='<yourAccessKey>'
);
INSERT INTO datahub_sink
SELECT
LOWER(name)
from datagen_source;