PiflowX组件-ReadFromKafka

ReadFromKafka组件

组件说明

从kafka中读取数据。

计算引擎

flink

有界性

Unbounded

组件分组

kafka

端口

Inport：默认端口

outport：默认端口

组件属性

名称	展示名称	默认值	允许值	是否必填	描述	例子
kafka_host	KAFKA_HOST	“”	无	是	逗号分隔的Kafka broker列表。	127.0.0.1:9092
topic	TOPIC	“”	无	否	读取数据的topic名。亦支持用分号间隔的topic列表，如 ‘topic-1;topic-2’。" "注意，‘topic’ 和 ‘topic-pattern’ 两个选项只能使用其中一个。	topic-1
topic_pattern	TOPIC_PATTERN	“”	无	否	匹配读取topic名称的正则表达式。在作业开始运行时，所有匹配该正则表达式的topic都将被Kafka consumer订阅。注意，‘topic’ 和 ‘topic-pattern’ 两个选项只能使用其中一个。	topic1_*
startup_mode	STARTUP_MODE	“”	Set(“earliest-offset”, “latest-offset”, “group-offsets”, “timestamp”, “specific-offsets”)	否	Kafka consumer 的启动模式。	earliest-offset
schema	SCHEMA	“”	无	否	Kafka消息的schema信息。	id:int,name:string,age:int
format	FORMAT	“”	Set(“json”, “csv”, “avro”, “parquet”, “orc”, “raw”, “protobuf”,“debezium-json”, “canal-json”, “maxwell-json”, “ogg-json”)	是	用来反序列化Kafka消息的格式。注意：该配置项和 ‘value.format’ 二者必需其一。	json
group	GROUP	“”	无	否	Kafka source的消费组id。如果未指定消费组ID，则会使用自动生成的"KafkaSource-{tableIdentifier}"作为消费组ID。	group_1
properties	PROPERTIES	“”	无	否	Kafka source连接器其他配置

ReadFromKafka示例配置

{"flow": {"name": "DataGenTest","uuid": "1234","stops": [{"uuid": "0000","name": "DataGen1","bundle": "cn.piflow.bundle.flink.common.DataGen","properties": {"schema": "[{\"filedName\":\"id\",\"filedType\":\"INT\",\"kind\":\"sequence\",\"start\":1,\"end\":10000},{\"filedName\":\"name\",\"filedType\":\"STRING\",\"kind\":\"random\",\"length\":15},{\"filedName\":\"age\",\"filedType\":\"INT\",\"kind\":\"random\",\"max\":100,\"min\":1}]","count": "100","ratio": "5"}},{"uuid": "1111","name": "WriteToKafka1","bundle": "cn.piflow.bundle.flink.kafka.WriteToKafka","properties": {"kafka_host": "hadoop01:9092","topic": "test","schema": "","format": "json","properties": "{}"}},{"uuid": "2222","name": "ReadFromKafka1","bundle": "cn.piflow.bundle.flink.kafka.ReadFromKafka","properties": {"kafka_host": "hadoop01:9092","topic": "test","group": "test","startup_mode": "earliest-offset","schema": "id:int,name:string,age:int","format": "json","properties": "{}"}},{"uuid": "3333","name": "ShowData1","bundle": "cn.piflow.bundle.flink.common.ShowData","properties": {"showNumber": "5000"}}],"paths": [{"from": "DataGen1","outport": "","inport": "","to": "WriteToKafka1"},{"from": "WriteToKafka1","outport": "","inport": "","to": "ReadFromKafka1"},{"from": "ReadFromKafka1","outport": "","inport": "","to": "ShowData1"}]}
}

示例说明

本示例演示了通过DataGen组件生成id，name，age3个字段100条数据，每秒生成5条数据，通过WriteToKafka组件将数据写入到kafka的test topic中，然后通过ReadFromKafka组件从test topic中读取数据，最后使用ShowData组件将数据打印在控制台。

字段描述

[{       "filedName": "id","filedType": "INT","kind": "sequence","start": 1,"end": 10000},{       "filedName": "name","filedType": "STRING","kind": "random","length": 15},{       "filedName": "age","filedType": "INT","kind": "random","max": 100,"min": 1} 
]

1.id字段

id字段类型为INT,使用sequence生成器，序列生成器的起始值为1，结束值为10000.

2.name字段

name字段类型为STRING,使用random生成器，生成字符长度为15。

3.age字段

age字段类型为INT，使用random生成器，随机生成器的最小值为1，最大值为100。