开源Kafka输入流

功能描述

创建source流从Kafka获取数据，作为作业的输入数据。

Apache Kafka是一个快速、可扩展的、高吞吐、可容错的分布式发布订阅消息系统，具有高吞吐量、内置分区、支持数据副本和容错的特性，适合在大规模消息处理场景中使用。

前提条件

• Kafka服务端的端口如果监听在hostname上，则需要将Kafka Broker节点的hostname和IP的对应关系添加到DLI队列中。Kafka Broker节点的hostname和IP请联系Kafka服务的部署人员。如何添加IP域名映射，请参考《数据湖探索用户指南》中“增强型跨源连接 > 修改主机信息”章节。
• Kafka是线下集群，需要通过增强型跨源连接功能将Flink作业与Kafka进行对接。且用户可以根据实际所需设置相应安全组规则。

  如何建立增强型跨源连接，请参考《数据湖探索用户指南》中“增强型跨源连接”章节。

  如何设置安全组规则，请参见《虚拟私有云用户指南》中“安全组”章节。

语法格式

1
2
3
4
5
6
7
8
9

CREATE SOURCE STREAM stream_id (attr_name attr_type (',' attr_name attr_type)* )
  WITH (
    type = "kafka",
    kafka_bootstrap_servers = "",
    kafka_group_id = "",
    kafka_topic = "",
    encode = "json",
    json_config=""
  );

关键字

注意事项

在创建Source Stream时可以指定时间模型以便在后续计算中使用，当前DLI支持Processing Time和Event Time两种时间模型，具体使用语法可以参考配置时间模型。

示例

• 从Kafka读取对象为test的topic。数据编码格式为json且不含嵌套，例如：{"attr1": "lilei", "attr2": 18}。

  1 2 3 4 5 6 7 8 9

  CREATE SOURCE STREAM kafka_source (name STRING, age int) WITH ( type = "kafka", kafka_bootstrap_servers = "ip1:port1,ip2:port2", kafka_group_id = "sourcegroup1", kafka_topic = "test", encode = "json", json_config = "name=attr1;age=attr2" );

• 从Kafka读取对象为test的topic。数据编码格式为json且包含嵌套。本示例使用了复杂数据类型ROW，ROW使用语法可以参考数据类型。

  测试数据参考如下：

  {
      "id":"1",
      "type2":"online",
      "data":{
          "patient_id":1234,
          "name":"bob1234"
      }
  }

  则对应建表语句示例为：

  CREATE SOURCE STREAM kafka_source 
  (
    id STRING,
    type2 STRING,
    data ROW<
      patient_id STRING, 
      name STRING>
  )
  WITH (
    type = "kafka",
    kafka_bootstrap_servers = "ip1:port1,ip2:port2", 
    kafka_group_id = "sourcegroup1", 
    kafka_topic = "test",
    encode = "json"
  );
  
  CREATE SINK STREAM kafka_sink 
  (
    id STRING,
    type2 STRING,
    patient_id STRING, 
    name STRING
  )
    WITH (
      type="kafka",
      kafka_bootstrap_servers =  "ip1:port1,ip2:port2",
      kafka_topic = "testsink",
      encode = "csv" 
    );
  
  INSERT INTO kafka_sink select id, type2, data.patient_id, data.name from kafka_source;