消息队列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消息的偏移量（offset）。
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的时间。

CTAS语句支持将消息队列Kafka，且format为JSON的表作为数据源。在数据同步过程中，如果某些字段并未出现在预定义的表结构中，Flink会尝试去自动推导该列的类型。如果自动推导的类型不能满足您的使用需求，您也可以通过辅助推导的方式对某些列的解析类型进行声明。关于类型推导和嵌套类型的处理方式，详情如下：

在类型推导过程中，Flink默认只会展开JSON文本中的第一层数据，根据其类型和数值，按照基本规则进行类型推导。当然您也可以在DDL中声明特定的列的解析类型以满足您的特殊需要，即辅助推导。基本规则和辅助推导详情如下：

• 基本规则
  类型映射基本规则如下表所示。

  JSON类型	Flink SQL类型
  BOOLEAN	BOOLEAN
  STRING	DATE、TIMESTAMP、TIMESTAMP_LTZ、TIME或STRING
  INT或LONG	BIGINT
  BIGINT	DECIMAL或STRING 说明 由于Flink中DECIMAL的类型是有精度限制的。因此，如果整数的实际取值超过了DECIMAL类型最大精度，Flink会自动推导其类型为STRING，避免精度的损失。
  FLOAT、DOUBLE或BIG DECIMAL	DOUBLE
  ARRAY	STRING
  OBJECT	STRING

  示例

  • JSON文本

    {
      "id": 101,
      "name": "VVP",
      "properties": {
        "owner": "阿里云",
        "engine": "Flink"
      }
        "type": ["大数据"]
    }

  • Flink写入到下游存储的表信息

    id	name	properties	type
    101	"VVP"	{ "owner": "阿里云", "engine": "Flink" }	["大数据"]

• 辅助推导
  如果您觉得以上基本规则不符合您实际需要，则您可以在源表的DDL中声明特定列的解析类型。通过该方式，Flink会优先使用您声明的列类型去解析目标字段。针对以下示例，Flink会使用DECIMAL的方式去解析price字段，而不是使用默认的基本规则将其转换为DOUBLE类型。

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

  但是，如果您在DDL中指定的类型和实际数据中的类型不一致时，则可以按照以下方式进行处理：

  • 在声明的类型比实际类型更宽泛时，以声明的类型自动去解析。

    例如，声明为DOUBLE，遇到的数据类型为BIGINT，则会以DOUBLE类型去解析。

  • 在实际的类型比声明的类型更为宽泛或者两种类型不兼容时，由于当前CTAS不支持类型变更，因此会报错提示您相关信息，您需要重新启动作业并声明准确的类型去解析数据。
    对于类型的宽泛的程度以及兼容性，可以总结为下图。上图表示越靠近根节点，其类型越宽泛。如果两个类型在不同的分支上，则表示这两个类型不兼容。

    说明

    • 不支持辅助推导复杂类型，包括ROW、ARRAY、MAP和MULTISET。
    • 对于复杂类型，Flink在默认情况下会处理为STRING。

通常，Kafka Topic中的JSON文本带有嵌套结构。如果您需要提取JSON文本中的嵌套列，则可以通过以下两种方式：

• 在源表DDL中声明'json.infer-schema.flatten-nested-columns.enable'='true'，来展开嵌套列中的所有元素至顶层。
  通过该方式，所有的嵌套列都会被依次展开。为了避免列名冲突，Flink采用索引到该列的路径作为展开后列名字。

  说明 目前不支持解决列名冲突。如果发生列名冲突，请在源表的DDL中声明json.ignore-parse-errors为true，来忽略存在冲突的数据。
  示例

  • JSON文本

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

  • Flink写入到下游存储的表信息

    neseted.inner.col
    true

• 在DDL中CTAS语法中添加计算列 `rowkey` AS JSON_VALUE(`properties`, `$.rowkey`)，来指定要展开的列。

  详情请参见示例四。

Kafka Consumer配置参数详情请参见Kafka官网消费者配置项列表。如果您还需要直接配置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.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="USERNAME" password="PASSWORD";'
);