# Upsert Kafka源表 #### 功能描述 Apache Kafka是一个快速、可扩展的、高吞吐、可容错的分布式发布订阅消息系统，具有高吞吐量、内置分区、支持数据副本和容错的特性，适合在大规模消息处理场景中使用。 作为 source，upsert-kafka 连接器生产changelog流，其中每条数据记录代表一个更新或删除事件。更准确地说，数据记录中的 value 被解释为同一 key 的最后一个 value 的 UPDATE，如果有这个 key（如果不存在相应的 key，则该更新被视为 INSERT）。用表来类比，changelog 流中的数据记录被解释为 UPSERT，也称为 INSERT/UPDATE，因为任何具有相同 key 的现有行都被覆盖。另外，value 为空的消息将会被视作为 DELETE 消息。 #### 前提条件 - 该场景作业需要运行在DLI的独享队列上，因此要与kafka集群建立增强型跨源连接，且用户可以根据实际所需设置相应安全组规则。 - 如何建立增强型跨源连接，请参考《数据湖探索用户指南》中[增强型跨源连接](https://support.huaweicloud.com/usermanual-dli/dli_01_0426.html)章节。 - 如何设置安全组规则，请参见[《虚拟私有云用户指南》](https://support.huaweicloud.com/usermanual-vpc/zh-cn_topic_0073379079.html)中"安全组"章节。 - Flink跨源开发场景中直接配置跨源认证信息存在密码泄露的风险，优先推荐您使用DLI提供的跨源认证。 跨源认证简介及操作方法请参考[跨源认证简介](https://support.huaweicloud.com/usermanual-dli/dli_01_0561.html)。 #### 注意事项 - 创建Flink OpenSource SQL作业时，在作业编辑界面的"运行参数"处，"Flink版本"需要选择"1.12"，勾选"保存作业日志"并设置保存作业日志的OBS桶，方便后续查看作业日志。 - Upsert Kafka 始终以upsert方式工作，并且需要在DDL中定义主键。在具有相同主键值的消息按序存储在同一个分区的前提下，在 changlog source 定义主键意味着 在物化后的 changelog 上主键具有唯一性。定义的主键将决定哪些字段出现在Kafka消息的key中。 - 由于该连接器以 upsert 的模式工作，该连接器作为 source 读入时，可以确保具有相同主键值下仅最后一条消息会生效。 - 数据类型的使用，请参考[Format](https://support.huaweicloud.com/sqlref-flink-dli/dli_08_0407.html)章节。 #### 语法格式 ``` create table kafkaSource( attr_name attr_type (',' attr_name attr_type)* (','PRIMARY KEY (attr_name, ...) NOT ENFORCED) ) with ( 'connector' = 'upsert-kafka', 'topic' = '', 'properties.bootstrap.servers' = '', 'key.format' = '', 'value.format' = '' ); ``` #### 参数说明 表1参数说明 | 参数 | 是否必选 | 默认参数 | 数据类型 | 说明 | |:---|:---|:---|:---|:---| | connector | 是 | 无 | String | connector类型，对于upsert kafka，需配置为'upsert-kafka'。 | | topic | 是 | 无 | String | Kafka topic名。 | | properties.bootstrap.servers | 是 | 无 | String | Kafka brokers地址，以逗号分隔。 | | key.format | 是 | 无 | String | 用于对Kafka消息中key部分序列化和反序列化的格式。key字段由PRIMARY KEY语法指定。支持的格式如下： - csv - json - avro 请参考[Format](https://support.huaweicloud.com/sqlref-flink-dli/dli_08_0407.html)页面以获取更多详细信息和格式参数。 | | key.fields-prefix | 否 | 无 | String | 为键格式的所有字段定义自定义前缀，以避免与值格式的字段发生名称冲突。 默认情况下，前缀为空。如果定义了自定义前缀，则表架构和'key.fields'都将使用前缀名称。在构造密钥格式的数据类型时，将删除前缀，并在密钥格式中使用无前缀的名称。请注意，此选项要求'value.fields-include' 必须设置为'EXCEPT_KEY'。 | | value.format | 是 | 无 | String | 用于对 Kafka消息中 value 部分序列化和反序列化的格式。支持的格式： - csv - json - avro 请参考[Format](https://support.huaweicloud.com/sqlref-flink-dli/dli_08_0407.html)页面以获取更多详细信息和格式参数。 | | value.fields-include | 是 | ALL | String | 控制哪些字段应该出现在值中。取值范围如下： - ALL：消息的value部分将包含schema的所有字段，包括定义中键的字段。 - EXCEPT_KEY：记录的value部分包含schema的所有内容，定义为主键的字段除外。 | | properties.\* | 否 | 无 | String | 该选项可以传递任意的Kafka参数。 "properties."后的后缀名必须匹配定义在 [kafka参数文档](https://kafka.apache.org/documentation/#configuration)中的参数名。 Flink会自动移除选项名中的 "properties." 前缀，并将转换后的键名以及值传入KafkaClient。 例如：您可以通过 'properties.allow.auto.create.topics' = 'false' 来禁止自动创建 topic。 但是'key.deserializer' 和 'value.deserializer' 是不允许通过该方式传递参数，因为Flink会重写这些参数的值。 | | ssl_auth_name | 否 | 无 | String | DLI侧创建的Kafka_SSL类型的跨源认证名称。Kafka配置SSL时使用该配置。 注意：若仅使用SSL类型，则需要同时配置'properties.security.protocol '= 'SSL'; 若使用SASL_SSL类型，则需要同时配置'properties.security.protocol' = 'SASL_SSL'、'properties.sasl.mechanism' = 'GSSAPI或者PLAIN'、'properties.sasl.jaas.config' = 'org.apache.kafka.common.security.plain.PlainLoginModule required username=\\"xxx\\" password=\\"xxx\\";' | | krb_auth_name | 否 | 无 | String | DLI侧创建的Kerberos类型的跨源认证名称。Kafka配置SASL认证时使用该配置。 注意：如果使用SASL_PLAINTEXT类型，且使用Kerberos认证，则需要同时配置'properties.sasl.mechanism' = 'GSSAPI'和'properties.security.protocol' = 'SASL_PLAINTEXT' | #### 示例 该示例是从Kafka数据源中读取数据，并写入Print到结果表中，其具体步骤如下： 1. 参考[增强型跨源连接](https://support.huaweicloud.com/usermanual-dli/dli_01_0426.html)，根据Kafka所在的虚拟私有云和子网创建相应的增强型跨源，并绑定所要使用的Flink弹性资源池。 2. 设置Kafka的安全组，添加入向规则使其对Flink的队列网段放通。参考[测试地址连通性](https://support.huaweicloud.com/usermanual-dli/dli_01_0489.html)根据Kafka的地址测试队列连通性。若能连通，则表示跨源已经绑定成功，否则表示未成功。 3. 参考[创建Flink OpenSource作业](https://support.huaweicloud.com/usermanual-dli/dli_01_0498.html)，创建flink opensource sql作业，输入以下作业脚本，提交运行作业。 注意：创建作业时，在作业编辑界面的"运行参数"处，"Flink版本"选择"1.12"，勾选"保存作业日志"并设置保存作业日志的OBS桶，方便后续查看作业日志。**如下脚本中的加粗参数请根据实际环境修改** 。 ``` CREATE TABLE upsertKafkaSource ( order_id string, order_channel string, order_time string, pay_amount double, real_pay double, pay_time string, user_id string, user_name string, area_id string, PRIMARY KEY (order_id) NOT ENFORCED ) WITH ( 'connector' = 'upsert-kafka', 'topic' = 'KafkaTopic', 'properties.bootstrap.servers' = 'KafkaAddress1:KafkaPort,KafkAddress2:KafkaPort', 'key.format' = 'csv', 'value.format' = 'json' ); CREATE TABLE printSink ( order_id string, order_channel string, order_time string, pay_amount double, real_pay double, pay_time string, user_id string, user_name string, area_id string, PRIMARY KEY (order_id) NOT ENFORCED ) WITH ( 'connector' = 'print' ); INSERT INTO printSink SELECT * FROM upsertKafkaSource; ``` 4. 向Kafka中的指定topic中插入如下数据（注意：**kafka插入数据时请指定key** ）。 ``` {"order_id":"202103251202020001", "order_channel":"miniAppShop", "order_time":"2021-03-25 12:02:02", "pay_amount":"60.00", "real_pay":"60.00", "pay_time":"2021-03-25 12:03:00", "user_id":"0002", "user_name":"Bob", "area_id":"330110"} {"order_id":"202103251505050001", "order_channel":"qqShop", "order_time":"2021-03-25 15:05:05", "pay_amount":"500.00", "real_pay":"400.00", "pay_time":"2021-03-25 15:10:00", "user_id":"0003", "user_name":"Cindy", "area_id":"330108"} {"order_id":"202103251202020001", "order_channel":"miniAppShop", "order_time":"2021-03-25 12:02:02", "pay_amount":"60.00", "real_pay":"60.00", "pay_time":"2021-03-25 12:03:00", "user_id":"0002", "user_name":"Bob", "area_id":"330110"} ``` 5. 用户可按下述操作查看输出结果： 1. 登录DLI管理控制台，选择"作业管理 \> Flink作业"。 2. 单击对应的Flink作业名称，选择"运行日志"，单击"OBS桶"，根据作业运行的日期，找到对应日志的文件夹。 3. 进入对应日期的文件夹后，找到名字中包含"taskmanager"的文件夹进入，下载获取taskmanager.out文件查看结果日志。 数据结果参考如下： ``` +I(202103251202020001,miniAppShop,2021-03-2512:02:02,60.0,60.0,2021-03-2512:03:00,0002,Bob,330110) +I(202103251505050001,qqShop,2021-03-2515:05:05,500.0,400.0,2021-03-2515:10:00,0003,Cindy,330108) -U(202103251202020001,miniAppShop,2021-03-2512:02:02,60.0,60.0,2021-03-2512:03:00,0002,Bob,330110) +U(202103251202020001,miniAppShop,2021-03-2512:02:02,60.0,60.0,2021-03-2512:03:00,0002,Bob,330110) ``` #### 常见问题 无