流计算 Oceanus 消息队列 Upsert Kafka

一、介绍

作为 Sink，Upsert Kafka 连接器可以消费 changelog 流。它会将 INSERT/UPDATE_AFTER 数据作为正常的 Kafka 消息写入，并将 DELETE 数据以 value 为空的 Kafka 消息写入（表示对应 key 的消息被删除）。Flink 将根据主键列的值对数据进行分区，从而保证主键上的消息有序，因此同一主键上的更新/删除消息将落在同一分区中。

二、版本说明

注意：说明：

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

三、DDL 定义

CREATE TABLE kafka_upsert_sink_table ( id INT, name STRING, PRIMARY KEY (id) NOT ENFORCED) WITH ( — 定义 Upsert Kafka 参数 'connector' = 'upsert-kafka', — 选择 connector 'topic' = 'topic', — 替换为你要写入的 Topic 'properties.bootstrap.servers' = '…', — 替换为你的 Kafka 连接地址 'key.format' = 'json', — 定义 key 数据格式 'value.format' = 'json' — 定义value 数据格式);

注意：说明：

注意：Upsert Kafka 确保在 DDL 中定义主键。

四、WITH 参数

五、代码示例

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' = '172.28.28.13:9092', — 替换为你的 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_upsert_sink_table ( id INT, name STRING, PRIMARY KEY (id) NOT ENFORCED) WITH ( — 定义 Upsert Kafka 参数 'connector' = 'upsert-kafka', — 选择 connector 'topic' = 'topic', — 替换为你要消费的 Topic 'properties.bootstrap.servers' = '…', — 替换为你的 Kafka 连接地址 'key.format' = 'json', — 定义 key 数据格式 'value.format' = 'json' — 定义value 数据格式);– 计算 pv、uv 并插入到 upsert-kafka sinkINSERT INTO kafka_upsert_sink_tableSELECT * FROM kafka_json_source_table;

六、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', …);

注意：说明：

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

七、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/172.28.28.51@EMR-OQPO48B9KVNO Timestamp Principal—- ——————- —————————————————— 2 08/09/2021 15:34:40 hadoop/172.28.28.51@EMR-OQPO48B9 2 08/09/2021 15:34:40 HTTP/172.28.28.51@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', …);

注意：说明：

注意：参数 properties.sasl.kerberos.service.name 的值必须与你选取的 principal 匹配，如果你选择的为 hadoop/${IP}@EMR-OQPO48B9，那么取值为 hadoop。

7. 作业 高级参数 配置。

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