流计算 Oceanus 消息队列 Kafka

一、介绍

Kafka 支持同一个 Topic 多分区读写，数据可以从多个分区读入，也可以写入到多个分区，以提供更高的吞吐量，减少数据倾斜和热点。

二、版本说明

注意：说明：

注意：自 Flink 1.17 版本起，社区已将 Kafka Connector 拆分为独立仓库进行维护，并采用了全新的版本号命名规则。

三、使用范围

Kafka 支持用作数据源表（Source），也可以作为 Tuple 数据流的目的表（Sink）。

Kafka 还可以与 Debezium、Canal 等联用，对 MySQL、PostgreSQL 等传统数据库的变更进行捕获和订阅，然后 Flink 即可对这些变更事件进行进一步的处理。

六、JSON 格式输入

CREATE TABLE `kafka_json_source_table` ( `id` INT, `name` STRING) WITH ( — 定义 Kafka 参数 'connector' = 'kafka', 'topic' = 'Data-Input', — 替换为你要消费的 Topic 'scan.startup.mode' = 'latest-offset', — 可以是 latest-offset / earliest-offset / specific-offsets / group-offsets / timestamp 的任何一种 'properties.bootstrap.servers' = '$IP:$PORT', — 替换为你的 Kafka 连接地址 'properties.group.id' = 'testGroup', — 必选参数, 一定要指定 Group ID — 定义数据格式 (JSON 格式) 'format' = 'json', 'json.fail-on-missing-field' = 'false', — 如果设置为 false, 则遇到缺失字段不会报错。 'json.ignore-parse-errors' = 'true' — 如果设置为 true，则忽略任何解析报错。);

七、CSV 格式输入

CREATE TABLE `kafka_csv_source_table` ( `id` INT, `name` STRING) WITH ( — 定义 Kafka 参数 'connector' = 'kafka', 'topic' = 'Data-Input', — 替换为你要消费的 Topic 'scan.startup.mode' = 'latest-offset', — 可以是 latest-offset / earliest-offset / specific-offsets / group-offsets / timestamp 的任何一种 'properties.bootstrap.servers' = '$IP:$PORT', — 替换为你的 Kafka 连接地址 'properties.group.id' = 'testGroup', — 必选参数, 一定要指定 Group ID — 定义数据格式 (CSV 格式) 'format' = 'csv');

八、Debezium 格式输入

CREATE TABLE `kafka_debezium_source_table` ( `id` INT, `name` STRING) WITH ( — 定义 Kafka 参数 'connector' = 'kafka', 'topic' = 'Data-Input', — 替换为你要消费的 Topic 'scan.startup.mode' = 'latest-offset', — 可以是 latest-offset / earliest-offset / specific-offsets / group-offsets / timestamp 的任何一种 'properties.bootstrap.servers' = '$IP:$PORT', — 替换为你的 Kafka 连接地址 'properties.group.id' = 'testGroup', — 必选参数, 一定要指定 Group ID — 定义数据格式 (Debezium 输出的 JSON 格式) 'format' = 'debezium-json');

九、Canal 格式输入

CREATE TABLE `kafka_source`( aid BIGINT COMMENT 'unique id', charname STRING, `ts` TIMESTAMP(6), origin_database STRING METADATA FROM 'value.database' VIRTUAL, origin_table STRING METADATA FROM 'value.table' VIRTUAL, origin_es TIMESTAMP(3) METADATA FROM 'value.event-timestamp' VIRTUAL, origin_type STRING METADATA FROM 'value.operation-type' VIRTUAL, `batch_id` BIGINT METADATA FROM 'value.batch-id' VIRTUAL, `is_ddl` BOOLEAN METADATA FROM 'value.is-ddl' VIRTUAL, origin_old ARRAY<MAP<STRING, STRING>> METADATA FROM 'value.update-before' VIRTUAL, `mysql_type` MAP<STRING, STRING> METADATA FROM 'value.mysql-type' VIRTUAL, origin_pk_names ARRAY<STRING> METADATA FROM 'value.pk-names' VIRTUAL, `sql` STRING METADATA FROM 'value.sql' VIRTUAL, origin_sql_type MAP<STRING, INT> METADATA FROM 'value.sql-type' VIRTUAL, `ingestion_ts` TIMESTAMP(3) METADATA FROM 'value.ingestion-timestamp' VIRTUAL) WITH ( 'connector' = 'kafka', — 注意选择对应的内置 Connector 'topic' = '$TOPIC', — 替换为你要消费的 Topic 'properties.bootstrap.servers' = '$IP:$PORT', — 替换为你的 Kafka 连接地址 'properties.group.id' = 'testGroup', — 必选参数, 一定要指定 Group ID 'scan.startup.mode' = 'latest-offset', 'scan.topic-partition-discovery.interval' = '300s', 'format' = 'canal-json', 'canal-json.ignore-parse-errors' = 'false', — 忽略 JSON 结构解析异常 'canal-json.source.append-mode' = 'true' — 仅支持Flink1.13及以上版本);

十、Protobuf 格式输入

1. 使用 Protobuf 格式时，在下面链接下载 flink-sql-protobuf的 jar，下载完成后 上传依赖。

2. 编写 .proto 文件

src/main/proto/user_event.proto

syntax = "proto3";package com.example.proto;option java_package = "com.example.proto";option java_outer_classname = "UserEventOuterClass";// Column names in the Flink DDL must match proto field names exactly// (case-sensitive, findFieldByName matching in PbSchemaValidationUtils).message UserEvent { int64 user_id = 1; // -> BIGINT string user_name = 2; // -> STRING int32 age = 3; // -> INT double score = 4; // -> DOUBLE bool vip = 5; // -> BOOLEAN}

3. 生成打包 Java 类

<project> <groupId>com.example</groupId> <artifactId>user-event-proto</artifactId> <version>1.0.0</version> <properties> <protobuf.version>3.21.7</protobuf.version> <maven.compiler.source>1.8</maven.compiler.source> <maven.compiler.target>1.8</maven.compiler.target> </properties> <dependencies> <dependency> <groupId>com.google.protobuf</groupId> <artifactId>protobuf-java</artifactId> <version>${protobuf.version}</version> <scope>provided</scope> </dependency> </dependencies> <build> <plugins> <plugin> <groupId>com.github.os72</groupId> <artifactId>protoc-jar-maven-plugin</artifactId> <version>3.11.4</version> <executions> <execution> <phase>generate-sources</phase> <goals><goal>run</goal></goals> <configuration> <protocArtifact>com.google.protobuf:protoc:${protobuf.version}</protocArtifact> <includeMavenTypes>direct</includeMavenTypes> <inputDirectories> <include>src/main/proto</include> </inputDirectories> <outputTargets> <outputTarget> <type>java</type> <addSources>none</addSources> <outputDirectory>target/generated-sources/proto</outputDirectory> </outputTarget> </outputTargets> </configuration> </execution> </executions> </plugin> <plugin> <groupId>org.codehaus.mojo</groupId> <artifactId>build-helper-maven-plugin</artifactId> <version>3.0.0</version> <executions> <execution> <phase>generate-sources</phase> <goals><goal>add-source</goal></goals> <configuration> <sources> <source>target/generated-sources/proto</source> </sources> </configuration> </execution> </executions> </plugin> </plugins> </build></project>

执行 `mvn clean package`，将产物 `user-event-proto-1.0.0.jar` 上传依赖 为作业的依赖 jar。

4. Protobuf 输入：从 Kafka 读取

CREATE TABLE kafka_protobuf_source ( `user_id` BIGINT, `user_name` STRING, `age` INT, `score` DOUBLE, `vip` BOOLEAN) WITH ( 'connector' = 'kafka', 'topic' = '$TOPIC', — 替换为你要消费的 Topic 'properties.bootstrap.servers' = '$IP:$PORT', — 替换为你的 Kafka 连接地址 'properties.group.id' = 'testGroup', — 替换为你的 Kafka 消费组 'scan.startup.mode' = 'latest-offset', 'format' = 'protobuf', 'protobuf.message-class-name' = 'com.example.proto.UserEventOuterClass$UserEvent', — 你定义的 protobuf 类 'protobuf.ignore-parse-errors' = 'true');

十二、JSON 格式输出

CREATE TABLE `kafka_json_sink_table` ( `id` INT, `name` STRING) WITH ( — 定义 Kafka 参数 'connector' = 'kafka', 'topic' = 'Data-Output', — 替换为你要写入的 Topic 'properties.bootstrap.servers' = '$IP:$PORT', — 替换为你的 Kafka 连接地址 — 定义数据格式 (JSON 格式) 'format' = 'json', 'json.fail-on-missing-field' = 'false', — 如果设置为 false, 则遇到缺失字段不会报错。 'json.ignore-parse-errors' = 'true' — 如果设置为 true，则忽略任何解析报错。);

十三、CSV 格式输出

CREATE TABLE `kafka_csv_sink_table` ( `id` INT, `name` STRING) WITH ( — 定义 Kafka 参数 'connector' = 'kafka', 'topic' = 'Data-Output', — 替换为你要写入的 Topic 'properties.bootstrap.servers' = '$IP:$PORT', — 替换为你的 Kafka 连接地址 — 定义数据格式 (CSV 格式) 'format' = 'csv');

十四、Canal 格式输出

CREATE TABLE `kafka_canal_json_sink_table`( aid BIGINT COMMENT 'unique id', charname STRING, `ts` TIMESTAMP(6), origin_database STRING METADATA FROM 'value.database', origin_table STRING METADATA FROM 'value.table', origin_ts TIMESTAMP(3) METADATA FROM 'value.event-timestamp', `type` STRING METADATA FROM 'value.operation-type', `batch_id` BIGINT METADATA FROM 'value.batch-id', `isDdl` BOOLEAN METADATA FROM 'value.is-ddl', `old` ARRAY<MAP<STRING, STRING>> METADATA FROM 'value.update-before', `mysql_type` MAP<STRING, STRING> METADATA FROM 'value.mysql-type', `pk_names` ARRAY<STRING> METADATA FROM 'value.pk-names', `sql` STRING METADATA FROM 'value.sql', `sql_type` MAP<STRING, INT> METADATA FROM 'value.sql-type', `ingestion_ts` TIMESTAMP(3) METADATA FROM 'value.ingestion-timestamp') WITH ( 'connector' = 'kafka', — 注意选择对应的内置 Connector 'topic' = '$TOPIC', — 替换为你要消费的 Topic 'properties.bootstrap.servers' = '$IP:$PORT', — 替换为你的 Kafka 连接地址 'properties.group.id' = 'testGroup', — 必选参数, 一定要指定 Group ID 'format' = 'canal-json');

十五、Protobuf 格式输出

1. Protobuf Jar 下载、.proto 编写、编译等准备工作，见上面的 Protobuf 格式输入 部分。

2. Protobuf 输出：向 Kafka 写入

CREATE TABLE `kafka_protobuf_sink` ( `user_id` BIGINT, `user_name` STRING, `age` INT, `score` DOUBLE, `vip` BOOLEAN) WITH ( 'connector' = 'kafka', 'topic' = '$TOPIC', — 替换为你要消费的 Topic 'properties.bootstrap.servers' = '$IP:$PORT', — 替换为你的 Kafka 连接地址 'format' = 'protobuf', 'protobuf.message-class-name' = 'com.example.proto.UserEventOuterClass$UserEvent' — 你定义的 protobuf 类);

十六、WITH 参数

注意：说明：

注意：Flink-1.13 版本可通过设置 'scan.partition-assignment.strategy' = 'ROUND_ROBIN' 将分区均匀分配到 Taskmanager。注意，该参数不可与

注意：'scan.partition-assignment.strategy' 参数同时使用，且在 Flink-1.14 及之后版本已经废弃。

十七、JSON 格式 WITH 参数

十八、CSV 格式 WITH 参数

十九、Debezium-json 格式 WITH 参数

20、Canal 格式 WITH 参数

21、Protobuf 格式 WITH 参数

22、元数据列

在源表和结果表中定义元数据列，可获取或写入 Kafka 消息的元数据。

CREATE TABLE `kafka_source_table`( –读取消息的topic作为`topic_name`字段 `topic_name` STRING NOT NULL METADATA FROM 'topic' VIRTUAL, –读取消息的offset作为`data_offset`字段 `data_offset` BIGINT NOT NULL METADATA FROM 'offset' VIRTUAL, –读取消息的时间戳作为`ts`字段 `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL, …) WITH ( 'connector' = 'kafka', …);CREATE TABLE `kafka_sink_table`( –`ts`字段的时间戳作为ProducerRecord时间戳写入Kafka `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp', …) WITH ( 'connector' = 'kafka', …);

以下列出了源表和结果表默认支持的元数据：

23、Canal 格式支持的元数据（仅支持 Flink1.13 版本集群）

以下元数据只能作为表定义中的只读（VIRTUAL）列，若元数据列与物理列冲突，元数据列可以使用 meta.列名：

25、JSON 格式使用示例

CREATE TABLE `kafka_json_source_table` ( `id` INT, `name` STRING) WITH ( — 定义 Kafka 参数 'connector' = 'kafka', 'topic' = 'Data-Input', — 替换为你要消费的 Topic 'scan.startup.mode' = 'latest-offset', — 可以是 latest-offset / earliest-offset / specific-offsets / group-offsets / timestamp 的任何一种 'properties.bootstrap.servers' = '$IP:$PORT', — 替换为你的 Kafka 连接地址 'properties.group.id' = 'testGroup', — 必选参数, 一定要指定 Group ID — 定义数据格式 (JSON 格式) 'format' = 'json', 'json.fail-on-missing-field' = 'false', — 如果设置为 false, 则遇到缺失字段不会报错。 'json.ignore-parse-errors' = 'true' — 如果设置为 true，则忽略任何解析报错。);CREATE TABLE `kafka_json_sink_table` ( `id` INT, `name` STRING) WITH ( — 定义 Kafka 参数 'connector' = 'kafka', 'topic' = 'Data-Output', — 替换为你要写入的 Topic 'properties.bootstrap.servers' = '$IP:$PORT', — 替换为你的 Kafka 连接地址 — 定义数据格式 (JSON 格式) 'format' = 'json', 'json.fail-on-missing-field' = 'false', — 如果设置为 false, 则遇到缺失字段不会报错。 'json.ignore-parse-errors' = 'true' — 如果设置为 true，则忽略任何解析报错。);insert into kafka_json_sink_table select * from kafka_json_source_table;

26、复杂嵌套 JSON 格式使用示例

·JSON 示例

{ "id": 1234567890, "name": "tom", "date": "2000-10-25", "obj": { "time1": "11:11:11", "str": "test", "lg": 1122334455 }, "arr": [ "aa", "bb", "cc", "dd" ], "rowinarr": [ { "f1": "f11", "f2": 111 }, { "f1": "f12", "f2": 222 } ], "time": "19:19:19", "timestamp": "1999-01-12 14:14:14", "map": { "flink": 123 }, "mapinmap": { "inner_map": { "key": 234 } }}

·Flink SQL

create table kafka_source( id BIGINT, name STRING, `date` DATE, obj ROW<time1 TIME,str STRING,lg BIGINT>, arr ARRAY<STRING>, rowinarr ARRAY<ROW<f1 STRING,f2 INT>>, `time` TIME, `timestamp` TIMESTAMP(3), `map` MAP<STRING,BIGINT>, mapinmap MAP<STRING,MAP<STRING,INT>>) with ( 'connector' = 'kafka', 'topic' = 'test-topic', 'scan.startup.mode' = 'latest-offset', 'properties.bootstrap.servers' = '$IP:$PORT', 'properties.group.id' = 'testGroup', 'format' = 'json');create table logger_sink ( id BIGINT, name STRING, `date` DATE, str STRING, arr ARRAY<STRING>, nameinarray STRING, rowinarr ARRAY<ROW<f1 STRING,f2 INT>>, f2 INT, `time` TIME, `timestamp` TIMESTAMP(3), `map` MAP<STRING,BIGINT>, flink BIGINT, mapinmap MAP<STRING,MAP<STRING,INT>>, `key` INT) with ( 'connector' = 'logger');insert into logger_sinkselect id, name, `date`, obj.str, arr, arr[4], rowinarr, rowinarr[1].f2, `time`, `timestamp`, `map`, `map`['flink'], mapinmap, mapinmap['inner_map']['key']from kafka_source;

·输出结果

+I(1234567890,tom,2000-10-25,test,[aa, bb, cc, dd],dd,[f11,111, f12,222],111,13:13:13,1999-01-12T14:14:14,{flink=123},123,{inner_map={key=234}},234)

注意：1. 各数据类型获取元素的方法：

注意：2. array 的起始下标从 1 开始。

27、JSON 数据类型映射

28、Canal 使用示例

CREATE TABLE `source`( `aid` BIGINT, `charname` STRING, `ts` TIMESTAMP(6), `database_name` STRING METADATA FROM 'value.database_name', `table_name` STRING METADATA FROM 'value.table_name', `op_ts` TIMESTAMP(3) METADATA FROM 'value.op_ts', `op_type` STRING METADATA FROM 'value.op_type', `batch_id` BIGINT METADATA FROM 'value.batch_id', `is_ddl` BOOLEAN METADATA FROM 'value.is_ddl', `update_before` ARRAY<MAP<STRING, STRING>> METADATA FROM 'value.update_before', `mysql_type` MAP<STRING, STRING> METADATA FROM 'value.mysql_type', `pk_names` ARRAY<STRING> METADATA FROM 'value.pk_names', `sql` STRING METADATA FROM 'value.sql', `sql_type` MAP<STRING, INT> METADATA FROM 'value.sql_type', `ingestion_ts` TIMESTAMP(3) METADATA FROM 'value.ts', PRIMARY KEY (`aid`) NOT ENFORCED) WITH ( 'connector' = 'mysql-cdc' , 'append-mode' = 'true', 'hostname' = '$IP', 'port' = '$PORT', 'username' = '$USERNAME', 'password' = '$PASSWORD', 'database-name' = 't_wr', 'table-name' = 't1', 'server-time-zone' = 'Asia/Shanghai', 'server-id' = '5500-5510');CREATE TABLE `kafka_canal_json_sink`( aid BIGINT COMMENT 'unique id', charname STRING, `ts` TIMESTAMP(6), origin_database STRING METADATA FROM 'value.database', origin_table STRING METADATA FROM 'value.table', origin_ts TIMESTAMP(3) METADATA FROM 'value.event-timestamp', `type` STRING METADATA FROM 'value.operation-type', `batch_id` BIGINT METADATA FROM 'value.batch-id', `isDdl` BOOLEAN METADATA FROM 'value.is-ddl', `old` ARRAY<MAP<STRING, STRING>> METADATA FROM 'value.update-before', `mysql_type` MAP<STRING, STRING> METADATA FROM 'value.mysql-type', `pk_names` ARRAY<STRING> METADATA FROM 'value.pk-names', `sql` STRING METADATA FROM 'value.sql', `sql_type` MAP<STRING, INT> METADATA FROM 'value.sql-type', `ingestion_ts` TIMESTAMP(3) METADATA FROM 'value.ingestion-timestamp') WITH ( 'connector' = 'kafka', 'topic' = 'TOPIC', — 替换为你要消费的 Topic 'properties.bootstrap.servers' = '$IP:$PORT', — 替换为你的 Kafka 连接地址 'format' = 'canal-json');insert into kafka_canal_json_sink select * from source;

CREATE TABLE `source`( `aid` BIGINT, `charname` STRING, `ts` TIMESTAMP(3), origin_database STRING METADATA FROM 'value.database' VIRTUAL, origin_table STRING METADATA FROM 'value.table' VIRTUAL, origin_es TIMESTAMP(3) METADATA FROM 'value.event-timestamp' VIRTUAL, origin_type STRING METADATA FROM 'value.operation-type' VIRTUAL, `batch_id` BIGINT METADATA FROM 'value.batch-id' VIRTUAL, `is_ddl` BOOLEAN METADATA FROM 'value.is-ddl' VIRTUAL, origin_old ARRAY<MAP<STRING, STRING>> METADATA FROM 'value.update-before' VIRTUAL, `mysql_type` MAP<STRING, STRING> METADATA FROM 'value.mysql-type' VIRTUAL, origin_pk_names ARRAY<STRING> METADATA FROM 'value.pk-names' VIRTUAL, `sql` STRING METADATA FROM 'value.sql' VIRTUAL, origin_sql_type MAP<STRING, INT> METADATA FROM 'value.sql-type' VIRTUAL, `ingestion_ts` TIMESTAMP(3) METADATA FROM 'value.ingestion-timestamp' VIRTUAL, WATERMARK FOR `origin_es` AS `origin_es` – INTERVAL '5' SECOND) WITH ( 'connector' = 'kafka', — 注意选择对应的内置 Connector 'topic' = '$TOPIC', — 替换为你要消费的 Topic 'properties.bootstrap.servers' = '$IP:PORT', — 替换为你的 Kafka 连接地址 'properties.group.id' = 'testGroup', — 必选参数, 一定要指定 Group ID 'scan.startup.mode' = 'latest-offset', 'scan.topic-partition-discovery.interval' = '300s', 'format' = 'canal-json', 'canal-json.source.append-mode' = 'true', — 仅支持Flink1.13 'canal-json.ignore-parse-errors' = 'false');CREATE TABLE `kafka_canal_json` ( `aid` BIGINT, `charname` STRING, `ts` TIMESTAMP(9), origin_database STRING, origin_table STRING, origin_es TIMESTAMP(9), origin_type STRING, `batch_id` BIGINT, `is_ddl` BOOLEAN, origin_old ARRAY<MAP<STRING, STRING>>, `mysql_type` MAP<STRING, STRING>, origin_pk_names ARRAY<STRING>, `sql` STRING, origin_sql_type MAP<STRING, INT>, `ingestion_ts` TIMESTAMP(9), dt STRING, hr STRING) PARTITIONED BY (dt, hr)with ( 'connector' = 'hive', 'hive-version' = '3.1.1', 'hive-database' = 'testdb', 'partition.time-extractor.timestamp-pattern'='$dt $hr:00:00', 'sink.partition-commit.trigger'='partition-time', 'sink.partition-commit.delay'='30 min', 'sink.partition-commit.policy.kind'='metastore,success-file');insert into kafka_canal_json select *,DATE_FORMAT(`origin_es`,'yyyy-MM-dd'),DATE_FORMAT(`origin_es`,'HH')from `source`;

29、Protobuf 使用示例

CREATE TABLE `datagen_source` ( `user_id` BIGINT, `user_name` STRING, `age` INT, `score` DOUBLE, `vip` BOOLEAN) WITH ( 'connector' = 'datagen', 'rows-per-second' = '10');CREATE TABLE `kafka_protobuf_sink` ( `user_id` BIGINT, `user_name` STRING, `age` INT, `score` DOUBLE, `vip` BOOLEAN) WITH ( 'connector' = 'kafka', 'topic' = '$TOPIC', — 替换为你要消费的 Topic 'properties.bootstrap.servers' = '$IP:$PORT', — 替换为你的 Kafka 连接地址 'format' = 'protobuf', 'protobuf.message-class-name' = 'com.example.proto.UserEventOuterClass$UserEvent' — 你定义的 protobuf 类);INSERT INTO `kafka_protobuf_sink` SELECT * FROM `datagen_source`;

CREATE TABLE `kafka_protobuf_source` ( `user_id` BIGINT, `user_name` STRING, `age` INT, `score` DOUBLE, `vip` BOOLEAN) WITH ( 'connector' = 'kafka', 'topic' = '$TOPIC', — 替换为你要消费的 Topic 'properties.bootstrap.servers' = '$IP:$PORT', — 替换为你的 Kafka 连接地址 'properties.group.id' = 'testGroup', — 替换为你的 Kafka 消费组 'scan.startup.mode' = 'latest-offset', 'format' = 'protobuf', 'protobuf.message-class-name' = 'com.example.proto.UserEventOuterClass$UserEvent', — 你定义的 protobuf 类 'protobuf.ignore-parse-errors' = 'true');CREATE TABLE `print_sink` ( `user_id` BIGINT, `user_name` STRING, `age` INT, `score` DOUBLE, `vip` BOOLEAN) WITH ( 'connector' = 'print');INSERT INTO `print_sink` SELECT * FROM `kafka_protobuf_source`;

31、SASL/PLAIN 用户名密码认证授权

1. 参考 消息队列 CKafka > 配置 ACL 策略，设置 Topic 按用户名密码访问的 SASL_PLAINTEXT 认证方式。

2. 参考 消息队列 CKafka > 添加路由策略，选择 SASL_PLAINTEXT 接入方式，并以该接入方式下的网络地址访问 Topic。

3. 作业配置 with 参数。

CREATE TABLE `YourTable` (…) WITH ( … 'properties.sasl.jaas.config' = 'org.apache.kafka.common.security.plain.PlainLoginModule required username="ckafka-xxxxxxxx#YourUserName" password="YourPassword";', 'properties.security.protocol' = 'SASL_PLAINTEXT', 'properties.sasl.mechanism' = 'PLAIN', …);

注意：1. username 是实例 ID + # + 刚配置的用户名，password 是刚配置的用户密码 。

注意：2. Flink 版本 SASL 认证配置 对应包：

32、SASL/GSSAPI Kerberos 认证授权

腾讯云 CKafka 暂时不支持 Kerberos 认证，你的自建 Kafka 如果开启了 Kerberos 认证，可参考如下步骤配置作业。

1. 获取你的自建 Kafka 集群的 Kerberos 配置文件，如果你基于腾讯云 EMR 集群自建，获取 krb5.conf、emr.keytab 文件，路径如下。

/etc/krb5.conf/var/krb5kdc/emr.keytab

2. 对步骤1中获取的文件打 jar 包。

jar cvf kafka-xxx.jar krb5.conf emr.keytab

3. 校验 jar 的结构（可以通过 vim 命令查看 vim kafka-xxx.jar），jar 里面包含如下信息，请确保文件不缺失且结构正确。

META-INF/META-INF/MANIFEST.MFemr.keytabkrb5.conf

4. 在 程序包管理 页面上传 jar 包，并在作业参数配置里引用该程序包。

5. 获取 Kerberos principal，用于作业 高级参数 配置。

klist -kt /var/krb5kdc/emr.keytab# 输出如下所示，选取第一个即可：hadoop/$IP@EMR-OQPO48B9, 注意换成你正确的 IP 地址KVNO Timestamp Principal—- ——————- —————————————————— 2 08/09/2021 15:34:40 hadoop/$IP@EMR-OQPO48B9 2 08/09/2021 15:34:40 HTTP/$IP@EMR-OQPO48B9 2 08/09/2021 15:34:40 hadoop/VM-28-51-centos@EMR-OQPO48B9 2 08/09/2021 15:34:40 HTTP/VM-28-51-centos@EMR-OQPO48B9

6. 作业 with 参数配置。

CREATE TABLE `YourTable` (…) WITH ( … 'properties.security.protocol' = 'SASL_PLAINTEXT', 'properties.sasl.mechanism' = 'GSSAPI', 'properties.sasl.kerberos.service.name' = 'hadoop', …);

7. 作业 高级参数 配置。

security.kerberos.login.principal: hadoop/$IP@EMR-4K3VR5FDsecurity.kerberos.login.keytab: emr.keytabsecurity.kerberos.login.conf: krb5.confsecurity.kerberos.login.contexts: KafkaClientfs.hdfs.hadoop.security.authentication: kerberos

注意：把 $IP 替换为机器正确的 IP 地址。

注意：历史 Oceanus 集群可能不支持该功能，你可通过 在线客服 联系我们升级集群管控服务，以支持 Kerberos 访问。

33、动态指定 Source 消费模式

当使用 Kafka Connector 作为 Source 时，我们可以在启动时自定义消费模式，目前支持的消费模式有: latest-offset/earliest-offset/group-offsets/timestamp 四种启动模式。 具体用法如下：

1. 在 SQL 代码中对应 With 参数部分加入如下代码定义：

'scan.startup.mode' = '#{CUSTOM_SCAN_MODE}', 'scan.startup.timestamp-millis' = '#{CUSTOM_TIMESTAMP}',

2. 加入之后, 在启动作业时会有选择框选择启动模式，选择“不使用任何快照，直接运行”，在“kafka consumer 启动模式”中可以选择所需启动模式。再进行启动即可实现指定模式启动。

35、Kafka source，Json 格式读取，反序列化报错

在使用 Kafka Source 并配置 'format' = 'json' 时，若消费到 Kafka 格式不规范的脏数据，极易引发 JsonParseException 或 NumberFormatException 等反序列化异常（一般报错会包含 "Failed to deserialize JSON"），进而导致作业报错中断。

针对这一场景，可通过在 Kafka Source 的 WITH 参数中配置 'json.ignore-parse-errors' = 'true' 来忽略 JSON 解析错误。

注意，该配置的作用范围不仅限于 Source 算子本身，也影响与其处于同一算子链（Operator Chaining）的下游算子（例如紧连的 Calc 算子）。这意味着，如果下游算子在数据处理过程中发生其他类型的报错，也会被一并静默忽略。如果需要观察到非 Source 的报错，可以在高级参数中加 pipeline.operator-chaining: false，把算子链打开，重启作业观察报错。重启作业后拓扑图会变化，需从 group-offsets 或者无状态重启。

36、Kafka source，Json 格式读取，没有数据流出

在使用 Kafka Source 并配置 'format' = 'json' 时，若发现作业已正常消费 Kafka 数据，但下游却无任何输出，这通常是由于消费到的数据均为脏数据，且 WITH 参数中配置了 'json.ignore-parse-errors' = 'true'，导致所有格式不合规的数据在解析失败后被静默丢弃。

针对此类问题，建议采取以下排查与验证策略：

1. 定位报错信息：可暂时移除 'json.ignore-parse-errors' = 'true' 参数并重启作业，通过观察系统抛出的具体反序列化异常，来进一步分析数据的格式问题。

2. 验证数据格式：若下游算子不便直接查询数据，可通过以下两种方式进行数据校验：

·登录 Kafka 控制台，直接查看对应 Topic 中的实际消息内容；

·将 Kafka Source 读取到的数据直接输出至 Logger，通过日志直观检查其是否符合规范的 JSON 格式，代码如下：

create table kafka_source( id BIGINT, name STRING) with ( 'connector' = 'kafka', 'topic' = 'test-topic', 'scan.startup.mode' = 'latest-offset', 'properties.bootstrap.servers' = '$IP:$PORT', 'properties.group.id' = 'testGroup', 'format' = 'json');create table logger_sink ( id BIGINT, name STRING) with ( 'connector' = 'logger', 'print-identifier' = 'DebugData');insert into logger_sinkselect id, namefrom kafka_source;

37、如何配置 Kafka 原生参数

通过 SQL 里面的 with 参数，支持配置 Kafka 原生参数，可配置 Kafka Consumer 参数（Source 可能用到），Kafka Producer 参数（SInk 可能用到），Kafka 连接参数（Source，Sink 都可能用到）等。

注意，所有 Kafka 原生参数必须添加 properties. 前缀，否则配置将无法生效。例如：

·bootstrap.servers 需写为 properties.bootstrap.servers，具体见上面示例。

·sasl.mechanism 需写为 properties.sasl.mechanism，具体见上面示例。

对于 DataStream 作业，配置 Kafka 原生参数，不需要加 properties. 前缀，例如：

KafkaSource.<String>builder() .setBootstrapServers("localhost:9092") … .setProperty("group.id", "test-group-1") .setProperty("security.protocol", "SASL_PLAINTEXT") .setProperty("sasl.mechanism", "PLAIN") …