消息队列Kafka源表

以下为创建Kafka源表的DDL示例，消息格式为CSV，包含5个字段。

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'
)

在实际使用中请根据实际情况配置字段名和WITH参数。

您可以在源表中定义元信息列，以获取Kafka消息的元信息。例如，当WITH参数中定义了多个topic时，如果在Kafka源表中定义了元信息列，那么Flink读取到的数据就会被标识是从哪个topic中读取的数据。

Key	数据类型	说明
topic	STRING NOT NULL METADATA VIRTUAL	Kafka消息所在的Topic名称。
partition	INT NOT NULL METADATA VIRTUAL	Kafka消息所在的Partition ID。
headers	MAP<STRING, BYTES> NOT NULL METADATA VIRTUAL	Kafka消息的消息头（header）。
leader-epoch	INT NOT NULL METADATA VIRTUAL	Kafka消息的Leader epoch。
offset	BIGINT NOT NULL METADATA VIRTUAL	Kafka消息的偏移量（offest）。
timestamp	TIMESTAMP(3) WITH LOCAL TIME ZONE NOT NULL METADATA VIRTUAL	Kafka消息的时间戳。
timestamp-type	STRING NOT NULL METADATA VIRTUAL	Kafka消息的时间戳类型： NoTimestampType：消息中没有定义时间戳。 CreateTime：消息产生的时间。 LogAppendTime：消息被添加到Kafka Broker的时间。

如果您还需要直接配置Connector使用的Kafka Consumer，可以在Kafka Consumer配置参数前添加 properties前缀，并将该Kafka Consumer配置信息追加至WITH参数。例如Kafka集群需要SASL（Simple Authentication and Security Layer）认证。

CREATE TABLE kafkaTable (
    ...
) WITH (
    ...
    '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";'
);

Kafka Consumer配置项详情请参见 Consumer Configs。

从名称为source的Topic中读取Kafka数据，再写入名称为sink的Topic，数据使用CSV格式。

CREATE TEMPORARY TABLE Kafka_source (
    id INT,
    name STRING,
    age INT
) WITH (
    'connector' = 'kafka',
    'topic' = 'source',
    'properties.bootstrap.servers' = '<yourKafkaBrokers>',
    'properties.group.id' = '<yourKafkaConsumerGroupId>',
    'format' = 'csv'
);

CREATE TEMPORARY TABLE Kafka_sink (
    id INT,
    name STRING,
    age INT
) WITH (
    'connector' = 'kafka',
    'topic' = 'sink',
    'properties.bootstrap.servers' = '<yourKafkaBrokers>',
    'properties.group.id' = '<yourKafkaConsumerGroupId>',
    'format' = 'csv'
);

INSERT INTO Kafka_sink SELECT id, name, age FROM Kafka_source;