Create a Message Queue for Apache Kafka source table

The following sample code shows how to use DDL statements to create a Message Queue for Apache Kafka source table. In this example, the message format is CSV and the table contains five fields.

CREATE TABLE kafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `category_id` BIGINT,
  `behavior` STRING,
  `topic` STRING METADATA VIRTUAL,
  `partition` BIGINT METADATA VIRTUAL
) WITH (
  'connector' = 'kafka',
  'topic' = 'my_excellent_topic',
  'properties.bootstrap.servers' = 'mykafka:9092',
  'properties.group.id' = 'my_excellent_group'
  'format' = 'csv',
  'scan.startup.mode' = 'earliest-offset'
)

You can configure the field names and parameters in the WITH clause based on your business requirements.

You can define metadata columns in the Message Queue for Apache Kafka source table to obtain the metadata of Message Queue for Apache Kafka messages. For example, if multiple topics are defined in the WITH clause and a metadata column is defined in the Message Queue for Apache Kafka source table, the topic from which Flink reads data is marked.

Key	Data type	Description
topic	STRING NOT NULL METADATA VIRTUAL	The name of the topic to which the Message Queue for Apache Kafka message belongs.
partition	INT NOT NULL METADATA VIRTUAL	The ID of the partition to which the Message Queue for Apache Kafka message belongs.
headers	MAP<STRING, BYTES> NOT NULL METADATA VIRTUAL	The headers of the Message Queue for Apache Kafka message.
leader-epoch	INT NOT NULL METADATA VIRTUAL	The leader epoch of the Message Queue for Apache Kafka message.
offset	BIGINT NOT NULL METADATA VIRTUAL	The offset of the Message Queue for Apache Kafka message.
timestamp	TIMESTAMP(3) WITH LOCAL TIME ZONE NOT NULL METADATA VIRTUAL	The timestamp of the Message Queue for Apache Kafka message.
timestamp-type	STRING NOT NULL METADATA VIRTUAL	The timestamp type of the Message Queue for Apache Kafka message. Valid values: NoTimestampType: indicates that no timestamp is defined in the message. CreateTime: indicates the time when the message was generated. LogAppendTime: indicates the time when the message was added to Kafka brokers.

The CREATE TABLE AS statement can be executed to synchronize data from a Message Queue for Apache Kafka source table that is in the JSON format. If specific fields do not exist in a predefined table schema during data synchronization, Flink attempts to automatically infer the data types of the columns. If the data types that are obtained after automatic type inference do not meet your business requirements, you can perform auxiliary type inference to declare the data types of the columns. The following section describes the type inference methods and the methods to process nested types.

By default, Flink displays only the first layer of data in the JSON text during type inference. Flink infers the SQL data types based on the JSON data types and values and the data type mappings. You can also declare the data types of specific columns in the DDL statement to meet your business requirements. This method is auxiliary type inference.

• Data type mappings
  The following table lists the mappings between JSON and Flink SQL data types.

  JSON data type	Flink SQL data type
  BOOLEAN	BOOLEAN
  STRING	DATE, TIMESTAMP, TIMESTAMP_LTZ, TIME, or STRING
  INT or LONG	BIGINT
  BIGINT	DECIMAL or STRING Note The precision of the DECIMAL data type in Flink is limited. If the value of an integer exceeds the maximum precision of the DECIMAL type, Flink considers the data type of the value as STRING to prevent the loss of precision.
  FLOAT, DOUBLE, or BIG DECIMAL	DOUBLE
  ARRAY	STRING
  OBJECT	STRING

  Example

  • JSON text

    {
      "id": 101,
      "name": "VVP",
      "properties": {
        "owner": "Alibaba Cloud",
        "engine": "Flink"
      }
        "type": ["Big data"]
    }

  • Information about the table to which Flink writes data

    id	name	properties	type
    101	"VVP"	{ "owner": "Alibaba Cloud", "engine": "Flink" }	["Big data"]

• Auxiliary type inference
  If the preceding rules do not meet your business requirements, you can declare the data type of a specific column in the DDL statement that is used to create the source table. If you use this method, Flink preferentially parses the field that you want to use as the data type of the column that is declared in the DDL statement. In this example, Flink parses the price field as the DECIMAL type instead of converting the field to the DOUBLE type based on the data type mappings.

  CREATE TABLE evolvingKafkaSource (
    price DECIMAL(18, 2)
  ) WITH (
    'connector' = 'kafka',
    'properties.bootstrap.servers' = 'localhost:9092',
    'topic' = 'evolving_kafka_demo',
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'json'
  );

  However, if the data type that you specify in the DDL statement is different from the actual data type, one of the following methods can be used to handle this issue:

  • If the declared data type of a column has a larger range than the actual data type, the field that you want to use is parsed based on the declared data type.

    For example, if the column is declared as the DOUBLE type and data in the column is of the BIGINT type, the data type is parsed as DOUBLE.

  • If the actual data type has a larger range than the declared data type or the actual data type is incompatible with the declared data type, you cannot execute the CREATE TABLE AS statement to synchronize data type changes. In this case, an error is returned. To resolve this issue, you must restart your job and declare a valid data type to parse data.
    The following figure shows the ranges of data types and their compatibility. In the preceding figure, the ranges of the data types that are closest to the root node are the largest. If two data types belong to different branches, the two data types are incompatible.

    Note

    • You cannot perform auxiliary type inference on complex data types, such as ROW, ARRAY, MAP, and MULTISET.
    • By default, Flink parses complex data types as STRING.

In most cases, JSON text in Kafka topics has a nested structure. If you want to extract nested columns from JSON text, you can use one of the following methods:

• Declare 'json.infer-schema.flatten-nested-columns.enable'='true' in the DDL statement that is used to create the source table to expand all elements in nested columns to the top level.
  In this way, all nested columns are expanded in sequence. To prevent column name conflicts, Flink uses the path that indexes the column as the name of the column after the column is expanded.

  Note No solution is provided to handle column name conflicts. If a column name conflict occurs, declare json.ignore-parse-errors as true in the DDL statement that is used to create the source table to ignore the data that has conflicts.
  Example

  • JSON text

    {
      "nested": {
        "inner": {
          "col": true
        }
      }
    }

  • Information about the table to which Flink writes data

    neseted.inner.col
    true

• Add `rowkey` AS JSON_VALUE(`properties`, `$.rowkey`) to the CREATE TABLE AS syntax in the DDL statement to add a computed column and specify the column that you want to expand.

  For more information, see Example 4.

For more information about how to configure the parameters of the Kafka consumer, see Consumer Configs in the Kafka 2.4 documentation. To directly configure the Kafka consumer used by the connector, add the properties prefix before the configuration parameters of the Kafka consumer and append the configurations to the parameters in the WITH clause. The following sample code shows that the Message Queue for Apache Kafka cluster requires Simple Authentication and Security Layer (SASL) authentication.

CREATE TABLE kafkaTable (
    ...
) WITH (
    ...
    'properties.security.protocol' = 'SASL_PLAINTEXT',
    'properties.sasl.mechanism' = 'PLAIN',
    'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="USERNAME" password="PASSWORD";'
);