更新时间:2024-06-06 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"}

前提条件

  1. 已创建DMS Kafka实例。

    具体步骤可参考:DMS Kafka入门指引

    创建DMS Kafka实例时,不能开启Kafka SASL_SSL

  2. 已创建DWS实例。

    具体创建DWS集群的操作可以参考创建DWS集群

整体作业开发流程

整体作业开发流程参考图1
图1 作业开发流程

步骤1:创建队列:创建DLI作业运行的队列。

步骤2:创建Kafka的Topic:创建Kafka生产消费数据的Topic。

步骤3:创建DWS数据库和表:创建DWS数据库和表信息。

步骤4:创建增强型跨源连接:DLI上创建连接Kafka和DWS的跨源连接,打通网络。

步骤5:运行作业:DLI上创建和运行Flink OpenSource作业。

步骤6:发送数据和查询结果:Kafka上发送流数据,在RDS上查看运行结果。

步骤1:创建队列

  1. 登录DLI管理控制台,在左侧导航栏单击“资源管理 > 队列管理”,可进入队列管理页面。
  2. 在队列管理界面,单击界面右上角的“购买队列”。
  3. 在“购买队列”界面,填写具体的队列配置参数,具体参数填写参考如下。
    • 计费模式:选择“按需计费”。
    • 区域和项目:保持默认值即可。
    • 名称:填写具体的队列名称。

      新建的队列名称,名称只能包含数字、英文字母和下划线,但不能是纯数字,且不能以下划线开头。长度限制:1~128个字符。

      队列名称不区分大小写,系统会自动转换为小写。

    • 类型:队列类型选择“通用队列”。“按需计费”时需要勾选“专属资源模式”
    • AZ策略、规格:保持默认即可。
    • 企业项目:当前选择为“default”。
    • 高级选项:选择“自定义”。
    • 网段:配置队列网段。例如,当前配置为10.0.0.0/16。

      队列的网段不能和DMS Kafka、RDS MySQL实例的子网网段有重合,否则后续创建跨源连接会失败。

    • 其他参数根据需要选择和配置。
  4. 参数配置完成后,单击“立即购买”,确认配置信息无误后,单击“提交”完成队列创建。

步骤2:创建Kafka的Topic

  1. 在Kafka管理控制台,选择“Kafka专享版”,单击对应的Kafka名称,进入到Kafka的基本信息页面。
  2. 单击“Topic管理 > 创建Topic”,创建一个Topic。Topic配置参数如下:
    • Topic名称。本示例输入为:testkafkatopic。
    • 分区数:1。
    • 副本数:1。

    其他参数保持默认即可。

步骤3:创建DWS数据库和表

  1. 参考使用gsql命令行客户端连接DWS集群连接已创建的DWS集群。
  2. 执行以下命令连接DWS集群的默认数据库“gaussdb”:
    gsql -d gaussdb -h DWS集群连接地址 -U dbadmin -p 8000 -W password -r
    • gaussdb:DWS集群默认数据库。
    • DWS集群连接地址:请参见获取集群连接地址进行获取。如果通过公网地址连接,请指定为集群“公网访问地址”或“公网访问域名”,如果通过内网地址连接,请指定为集群“内网访问地址”或“内网访问域名”。如果通过弹性负载均衡连接,请指定为“弹性负载均衡地址”。
    • dbadmin:创建集群时设置的默认管理员用户名。
    • password :默认管理员用户的密码。
  3. 在命令行窗口输入以下命令创建数据库“testdwsdb”。
    CREATE DATABASE testdwsdb;
  4. 执行以下命令,退出gaussdb数据库,连接新创建的数据库“testdwsdb”。
    \q
    gsql -d testdwsdb -h DWS集群连接地址 -U dbadmin -p 8000 -W password -r
  5. 执行以下命令创建表。
    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的增强型跨源连接
    1. 在Kafka管理控制台,选择“Kafka专享版”,单击对应的Kafka名称,进入到Kafka的基本信息页面。
    2. 在“连接信息”中获取该Kafka的“内网连接地址”,在“基本信息”的“网络”中获取获取该实例的“虚拟私有云”和“子网”信息,方便后续操作步骤使用。
    3. 单击“网络”中的安全组名称,在“入方向规则”中添加放通队列网段的规则。例如,本示例队列网段为“10.0.0.0/16”,则规则添加为:优先级选为:1,策略选为:允许,协议选择:TCP,端口值不填,类型:IPv4,源地址为:10.0.0.0/16,单击“确定”完成安全组规则添加。
    4. 登录DLI管理控制台,在左侧导航栏单击“跨源管理”,在跨源管理界面,单击“增强型跨源”,单击“创建”。
    5. 在增强型跨源创建界面,配置具体的跨源连接参数。具体参考如下。
      • 连接名称:设置具体的增强型跨源名称。本示例输入为:dli_kafka。
      • 弹性资源池:选择步骤1:创建队列中已经创建的队列名。
      • 虚拟私有云:选择Kafka的虚拟私有云。
      • 子网:选择Kafka的子网。
      • 其他参数可以根据需要选择配置。

      参数配置完成后,单击“确定”完成增强型跨源配置。单击创建的跨源连接名称,查看跨源连接的连接状态,等待连接状态为:“已激活”后可以进行后续步骤。

    6. 单击“队列管理”,选择操作的队列,本示例为步骤1:创建队列中创建的队列,在操作列,单击“更多 > 测试地址连通性”。
    7. 在“测试连通性”界面,根据2中获取的Kafka连接信息,地址栏输入“Kafka内网地址:Kafka数据库端口”,单击“测试”测试DLI到Kafka网络是否可达。
  • 创建DLI连接DWS的增强型跨源连接
    1. 在DWS管理控制台,选择“集群管理”,单击已创建的DWS集群名称,进入到DWS的基本信息页面。
    2. 在“基本信息”的“数据库属性”中获取该实例的“内网IP”、“端口”,“基本信息”页面的“网络”中获取“虚拟私有云”和“子网”信息,方便后续操作步骤使用。
    3. 单击“连接信息”中的安全组名称,在“入方向规则”中添加放通队列网段的规则。例如,本示例队列网段为“10.0.0.0/16”,则规则添加为:优先级选为:1,策略选为:允许,协议选择:TCP,端口值不填,类型:IPv4,源地址为:10.0.0.0/16,单击“确定”完成安全组规则添加。
    4. Kafka和DWS实例属于同一VPC和子网下?
      1. 是,执行7。Kafka和DWS实例在同一VPC和子网,不用再重复创建增强型跨源连接。
      2. 否,执行5。Kafka和DWS实例分别在两个VPC和子网下,则要分别创建增强型跨源连接打通网络。
    5. 登录DLI管理控制台,在左侧导航栏单击“跨源管理”,在跨源管理界面,单击“增强型跨源”,单击“创建”。
    6. 在增强型跨源创建界面,配置具体的跨源连接参数。具体参考如下。
      • 连接名称:设置具体的增强型跨源名称。本示例输入为:dli_dws。
      • 弹性资源池:选择步骤1:创建队列中已经创建的队列名。
      • 虚拟私有云:选择DWS的虚拟私有云。
      • 子网:选择DWS的子网。
      • 其他参数可以根据需要选择配置。

      参数配置完成后,单击“确定”完成增强型跨源配置。单击创建的跨源连接名称,查看跨源连接的连接状态,等待连接状态为:“已激活”后可以进行后续步骤。

    7. 单击“队列管理”,选择操作的队列,本示例为步骤1:创建队列中创建的队列,在操作列,单击“更多 > 测试地址连通性”。
    8. 在“测试连通性”界面,根据2中获取的DWS连接信息,地址栏输入“DWS内网IP:DWS端口”,单击“测试”测试DLI到DWS网络是否可达。

步骤5:运行作业

  1. 在DLI管理控制台,单击“作业管理 > Flink作业”,在Flink作业管理界面,单击“创建作业”。
  2. 在创建队列界面,类型选择“Flink OpenSource SQL”,名称填写为:FlinkKafkaDWS。单击“确定”,跳转到Flink作业编辑界面。
  3. 在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;
  4. 单击“语义校验”确保SQL语义校验成功。单击“保存”,保存作业。单击“启动”,启动作业,确认作业参数信息,单击“立即启动”开始执行作业。等待作业运行状态变为“运行中”。

步骤6:发送数据和查询结果

  1. 使用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"}
  2. 连接已创建的DWS集群。

    具体操作请参考使用gsql命令行客户端连接DWS集群

  3. 执行以下命令连接DWS集群的默认数据库“testdwsdb”:
    gsql -d testdwsdb -h DWS集群连接地址 -U dbadmin -p 8000 -W password -r
  4. 查询DWS的表数据。
    select * from test.qualified_cars;
    查询结果参考如下:
    car_id  car_owner  car_age  average_speed  total_miles
    3027      lilei     7           76.0       15000.0