MRS Kafka输入流

功能描述

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

Apache Kafka是一个快速、可扩展的、高吞吐、可容错的分布式发布订阅消息系统，具有高吞吐量、内置分区、支持数据副本和容错的特性，适合在大规模消息处理场景中使用。MRS基于Apache Kafka在平台部署并托管了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

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"
  );

关键字

注意事项

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

示例

• 从Kafka名称为test的topic中读取数据。

  1 2 3 4 5 6 7 8 9 10 11

  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" );

• 从Kafka读取对象为test的topic，使用json_config将json数据和表字段对应。

  数据编码格式为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" );