更新时间：2024-12-05 GMT+08:00
从Kafka读取数据写入到DWS
本指导仅适用于Flink 1.12版本。
场景描述
该场景为对汽车驾驶的实时数据信息进行分析，将满足特定条件的数据结果进行汇总。汽车驾驶的实时数据信息为数据源发送到Kafka中，再将Kafka数据的分析结果输出到DWS中。
例如，输入如下样例数据：
{"car_id":"3027", "car_owner":"lilei", "car_age":"7", "average_speed":"76", "total_miles":"15000"}
{"car_id":"3028", "car_owner":"hanmeimei", "car_age":"6", "average_speed":"92", "total_miles":"17000"}
{"car_id":"3029", "car_owner":"Ann", "car_age":"10", "average_speed":"81", "total_miles":"230000"}
预期输出：average_speed <= 90 和 total_miles <= 200000的车辆，即：
{"car_id":"3027", "car_owner":"lilei", "car_age":"7", "average_speed":"76", "total_miles":"15000"}
前提条件
- 已创建DMS Kafka实例。
具体步骤可参考：DMS Kafka入门指引。
创建DMS Kafka实例时，不能开启Kafka SASL_SSL。
- 已创建DWS实例。
具体创建DWS集群的操作可以参考创建DWS集群。
整体作业开发流程
整体作业开发流程参考图1。
步骤1：创建队列：创建DLI作业运行的队列。
步骤2：创建Kafka的Topic：创建Kafka生产消费数据的Topic。
步骤3：创建DWS数据库和表：创建DWS数据库和表信息。
步骤4：创建增强型跨源连接：DLI上创建连接Kafka和DWS的跨源连接，打通网络。
步骤5：运行作业：DLI上创建和运行Flink OpenSource作业。
步骤6：发送数据和查询结果：Kafka上发送流数据，在RDS上查看运行结果。
步骤1：创建队列
- 登录DLI管理控制台，在左侧导航栏单击“资源管理 > 队列管理”，可进入队列管理页面。
- 在队列管理界面，单击界面右上角的“购买队列”。
- 在“购买队列”界面，填写具体的队列配置参数，具体参数填写参考如下。
- 计费模式：选择“包年/包月”或“按需计费”。本例选择“按需计费”。
- 区域和项目：保持默认值即可。
- 名称：填写具体的队列名称。
新建的队列名称，名称只能包含数字、英文字母和下划线，但不能是纯数字，且不能以下划线开头。长度限制：1~128个字符。
队列名称不区分大小写，系统会自动转换为小写。
- 类型：队列类型选择“通用队列”。“按需计费”时需要勾选“专属资源模式”。
- AZ策略、CPU架构、规格：保持默认即可。
- 企业项目：当前选择为“default”。
- 高级选项：选择“自定义”。
- 网段：配置队列网段。例如，当前配置为10.0.0.0/16。
队列的网段不能和DMS Kafka、RDS MySQL实例的子网网段有重合，否则后续创建跨源连接会失败。
- 其他参数根据需要选择和配置。
- 参数配置完成后，单击“立即购买”，确认配置信息无误后，单击“提交”完成队列创建。
步骤2：创建Kafka的Topic
- 在Kafka管理控制台，选择“Kafka专享版”，单击对应的Kafka名称，进入到Kafka的基本信息页面。
- 单击“Topic管理 > 创建Topic”，创建一个Topic。Topic配置参数如下：
- Topic名称。本示例输入为：testkafkatopic。
- 分区数：1。
- 副本数：1。
其他参数保持默认即可。
步骤3：创建DWS数据库和表
- 参考使用gsql命令行客户端连接DWS集群连接已创建的DWS集群。
- 执行以下命令连接DWS集群的默认数据库“gaussdb”：
gsql -d gaussdb -h DWS集群连接地址 -U dbadmin -p 8000 -W password -r
- gaussdb：DWS集群默认数据库。
- DWS集群连接地址：请参见获取集群连接地址进行获取。如果通过公网地址连接，请指定为集群“公网访问地址”或“公网访问域名”，如果通过内网地址连接，请指定为集群“内网访问地址”或“内网访问域名”。如果通过弹性负载均衡连接，请指定为“弹性负载均衡地址”。
- dbadmin：创建集群时设置的默认管理员用户名。
- password ：默认管理员用户的密码。
- 在命令行窗口输入以下命令创建数据库“testdwsdb”。
CREATE DATABASE testdwsdb;
- 执行以下命令，退出gaussdb数据库，连接新创建的数据库“testdwsdb”。
\q gsql -d testdwsdb -h DWS集群连接地址 -U dbadmin -p 8000 -W password -r
- 执行以下命令创建表。
create schema test; set current_schema= test; drop table if exists qualified_cars; CREATE TABLE qualified_cars ( car_id VARCHAR, car_owner VARCHAR, car_age INTEGER , average_speed FLOAT8, total_miles FLOAT8 );
步骤4：创建增强型跨源连接
- 创建DLI连接Kafka的增强型跨源连接
- 在Kafka管理控制台，选择“Kafka专享版”，单击对应的Kafka名称，进入到Kafka的基本信息页面。
- 在“连接信息”中获取该Kafka的“内网连接地址”，在“基本信息”的“网络”中获取获取该实例的“虚拟私有云”和“子网”信息，方便后续操作步骤使用。
- 单击“网络”中的安全组名称，在“入方向规则”中添加放通队列网段的规则。例如，本示例队列网段为“10.0.0.0/16”，则规则添加为：优先级选择：1，策略选择：允许，协议选择：TCP，端口值不填，类型：IPv4，源地址为：10.0.0.0/16，单击“确定”完成安全组规则添加。
- 登录DLI管理控制台，在左侧导航栏单击“跨源管理”，在跨源管理界面，单击“增强型跨源”，单击“创建”。
- 在增强型跨源创建界面，配置具体的跨源连接参数。具体参考如下。
- 连接名称：设置具体的增强型跨源名称。本示例输入为：dli_kafka。
- 弹性资源池：选择步骤1：创建队列中已经创建的队列名。
- 虚拟私有云：选择Kafka的虚拟私有云。
- 子网：选择Kafka的子网。
- 其他参数可以根据需要选择配置。
参数配置完成后，单击“确定”完成增强型跨源配置。单击创建的跨源连接名称，查看跨源连接的连接状态，等待连接状态为：“已激活”后可以进行后续步骤。
- 单击“队列管理”，选择操作的队列，本示例为步骤1：创建队列中创建的队列，在操作列，单击“更多 > 测试地址连通性”。
- 在“测试连通性”界面，根据2中获取的Kafka连接信息，地址栏输入“Kafka内网地址:Kafka数据库端口”，单击“测试”测试DLI到Kafka网络是否可达。
- 创建DLI连接DWS的增强型跨源连接
- 在DWS管理控制台，选择“集群管理”，单击已创建的DWS集群名称，进入到DWS的基本信息页面。
- 在“基本信息”的“数据库属性”中获取该实例的“内网IP”、“端口”，“基本信息”页面的“网络”中获取“虚拟私有云”和“子网”信息，方便后续操作步骤使用。
- 单击“连接信息”中的安全组名称，在“入方向规则”中添加放通队列网段的规则。例如，本示例队列网段为“10.0.0.0/16”，则规则添加为：优先级选择：1，策略选择：允许，协议选择：TCP，端口值不填，类型：IPv4，源地址为：10.0.0.0/16，单击“确定”完成安全组规则添加。
- Kafka和DWS实例属于同一VPC和子网下？
- 登录DLI管理控制台，在左侧导航栏单击“跨源管理”，在跨源管理界面，单击“增强型跨源”，单击“创建”。
- 在增强型跨源创建界面，配置具体的跨源连接参数。具体参考如下。
- 连接名称：设置具体的增强型跨源名称。本示例输入为：dli_dws。
- 弹性资源池：选择步骤1：创建队列中已经创建的队列名。
- 虚拟私有云：选择DWS的虚拟私有云。
- 子网：选择DWS的子网。
- 其他参数可以根据需要选择配置。
参数配置完成后，单击“确定”完成增强型跨源配置。单击创建的跨源连接名称，查看跨源连接的连接状态，等待连接状态为：“已激活”后可以进行后续步骤。
- 单击“队列管理”，选择操作的队列，本示例为步骤1：创建队列中创建的队列，在操作列，单击“更多 > 测试地址连通性”。
- 在“测试连通性”界面，根据2中获取的DWS连接信息，地址栏输入“DWS内网IP:DWS端口”，单击“测试”测试DLI到DWS网络是否可达。
步骤5：运行作业
- 在DLI管理控制台，单击“作业管理 > Flink作业”，在Flink作业管理界面，单击“创建作业”。
- 在创建队列界面，类型选择“Flink OpenSource SQL”，名称填写为：FlinkKafkaDWS。单击“确定”，跳转到Flink作业编辑界面。
- 在Flink OpenSource SQL作业编辑界面，配置如下参数，其他参数默认即可。
- 所属队列：选择步骤1：创建队列中创建的队列。
- Flink版本：选择1.12。
- 保存作业日志：勾选。
- OBS桶：选择保存作业日志的OBS桶，根据提示进行OBS桶权限授权。
- 开启Checkpoint：勾选。
- Flink作业编辑框中输入具体的作业SQL，本示例作业参考如下。SQL中加粗的参数需要根据实际情况修改。
本示例使用的Flink版本为1.12，故Flink OpenSource SQL语法也是1.12。本示例数据源是Kafka，写入结果数据到DWS。
请参考Flink OpenSource SQL 1.12创建Kafka源表和Flink OpenSource SQL 1.12创建DWS结果表。
create table car_infos( car_id STRING, car_owner STRING, car_age INT, average_speed DOUBLE, total_miles DOUBLE ) with ( "connector" = "kafka", "properties.bootstrap.servers" = "10.128.0.120:9092,10.128.0.89:9092,10.128.0.83:9092",--替换为kafka的内网连接地址和端口 "properties.group.id" = "click", "topic" = "testkafkatopic",--创建的Kafka的Topic "format" = "json", "scan.startup.mode" = "latest-offset" ); create table qualified_cars ( car_id STRING, car_owner STRING, car_age INT, average_speed DOUBLE, total_miles DOUBLE ) WITH ( 'connector' = 'gaussdb', 'driver' = 'com.huawei.gauss200.jdbc.Driver', 'url' = 'jdbc:gaussdb://192.168.168.16:8000/testdwsdb', ---192.168.168.16:8000替换为DWS的内网IP和端口，testdwsdb为创建的DWS数据库名 'table-name' = 'test\".\"qualified_cars', ---test为创建的DWS表的schema，qualified_cars为对应的DWS表名 'pwd_auth_name'= 'xxxxx', --DLI侧创建的Password类型的跨源认证名称。使用跨源认证则无需在作业中配置账号和密码。 'write.mode' = 'insert' ); /** 将合格车辆信息输出 **/ INSERT INTO qualified_cars SELECT * FROM car_infos where average_speed <= 90 and total_miles <= 200000;
- 单击“语义校验”确保SQL语义校验成功。单击“保存”，保存作业。单击“启动”，启动作业，确认作业参数信息，单击“立即启动”开始执行作业。等待作业运行状态变为“运行中”。
步骤6：发送数据和查询结果
- 使用Kafka客户端向步骤2：创建Kafka的Topic中的Topic发送数据，模拟实时数据流。
Kafka生产和发送数据的方法请参考DMS - 连接实例生产消费信息。
发送样例数据如下：
{"car_id":"3027", "car_owner":"lilei", "car_age":"7", "average_speed":"76", "total_miles":"15000"} {"car_id":"3028", "car_owner":"hanmeimei", "car_age":"6", "average_speed":"92", "total_miles":"17000"} {"car_id":"3029", "car_owner":"Ann", "car_age":"10", "average_speed":"81", "total_miles":"230000"} - 连接已创建的DWS集群。
具体操作请参考使用gsql命令行客户端连接DWS集群。
- 执行以下命令连接DWS集群的默认数据库“testdwsdb”：
gsql -d testdwsdb -h DWS集群连接地址 -U dbadmin -p 8000 -W password -r
- 查询DWS的表数据。
select * from test.qualified_cars;
查询结果参考如下：car_id car_owner car_age average_speed total_miles 3027 lilei 7 76.0 15000.0
