数据湖探索 DLI-Kafka结果表:示例（适用于Kafka集群未开启SASL_SSL场景）

时间：2025-02-12 15:00:46

数据湖探索 DLI

示例（适用于Kafka集群未开启SASL_SSL场景）

该示例是从Kafka的一个topic中读取数据，并使用Kafka结果表将数据写入到kafka的另一个topic中。

参考增强型跨源连接，根据Kafka所在的虚拟私有云和子网创建相应的增强型跨源，并绑定所要使用的Flink弹性资源池。
设置Kafka的安全组，添加入向规则使其对Flink的队列网段放通。参考测试地址连通性根据Kafka的地址测试队列连通性。若能连通，则表示跨源已经绑定成功，否则表示未成功。

参考创建Flink OpenSource作业，创建flink opensource sql作业，输入以下作业脚本，提交运行作业。

注意：创建作业时，在作业编辑界面的“运行参数”处，“Flink版本”选择“1.12”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。如下脚本中的加粗参数请根据实际环境修改。

CREATE TABLE kafkaSource (  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) WITH (  'connector' = 'kafka',  'topic' = 'KafkaTopic',  'properties.bootstrap.servers' = 'KafkaAddress1:KafkaPort,KafkaAddress2:KafkaPort',  'properties.group.id' = 'GroupId',  'scan.startup.mode' = 'latest-offset',  "format" = "json");CREATE TABLE kafkaSink (  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) WITH (  'connector' = 'kafka',  'topic' = 'KafkaSinkTopic',  'properties.bootstrap.servers' = 'KafkaAddress1:KafkaPort,KafkaAddress2:KafkaPort',  "format" = "json");insert into kafkaSink select * from kafkaSource;

连接Kafka集群，向Kafka的source topic中插入如下测试数据：

{"order_id":"202103241000000001","order_channel":"webShop","order_time":"2021-03-24 10:00:00","pay_amount":100.0,"real_pay":100.0,"pay_time":"2021-03-24 10:02:03","user_id":"0001","user_name":"Alice","area_id":"330106"}{"order_id":"202103241606060001","order_channel":"appShop","order_time":"2021-03-24 16:06:06","pay_amount":200.0,"real_pay":180.0,"pay_time":"2021-03-24 16:10:06","user_id":"0001","user_name":"Alice","area_id":"330106"}

连接Kafka集群，在Kafka的sink topic读取数据，参考如下：

{"order_id":"202103241000000001","order_channel":"webShop","order_time":"2021-03-24 10:00:00","pay_amount":100.0,"real_pay":100.0,"pay_time":"2021-03-24 10:02:03","user_id":"0001","user_name":"Alice","area_id":"330106"}{"order_id":"202103241606060001","order_channel":"appShop","order_time":"2021-03-24 16:06:06","pay_amount":200.0,"real_pay":180.0,"pay_time":"2021-03-24 16:10:06","user_id":"0001","user_name":"Alice","area_id":"330106"}

上一篇：数据湖探索 DLI-Kafka结果表:语法格式

下一篇：数据湖探索 DLI-Kafka结果表:语法格式