PiflowX组件-ReadFromKafka

ReadFromKafka组件

组件说明

从kafka中读取数据。

计算引擎

flink

有界性

Unbounded

组件分组

kafka

端口

Inport:默认端口

outport:默认端口

组件属性

名称展示名称默认值允许值是否必填描述例子
kafka_hostKAFKA_HOST“”逗号分隔的Kafka broker列表。127.0.0.1:9092
topicTOPIC“”读取数据的topic名。亦支持用分号间隔的topic列表,如 ‘topic-1;topic-2’。" "注意,‘topic’ 和 ‘topic-pattern’ 两个选项只能使用其中一个。topic-1
topic_patternTOPIC_PATTERN“”匹配读取topic名称的正则表达式。在作业开始运行时,所有匹配该正则表达式的topic都将被Kafka consumer订阅。注意,‘topic’ 和 ‘topic-pattern’ 两个选项只能使用其中一个。topic1_*
startup_modeSTARTUP_MODE“”Set(“earliest-offset”, “latest-offset”, “group-offsets”, “timestamp”, “specific-offsets”)Kafka consumer 的启动模式。earliest-offset
schemaSCHEMA“”Kafka消息的schema信息。id:int,name:string,age:int
formatFORMAT“”Set(“json”, “csv”, “avro”, “parquet”, “orc”, “raw”, “protobuf”,“debezium-json”, “canal-json”, “maxwell-json”, “ogg-json”)用来反序列化Kafka消息的格式。注意:该配置项和 ‘value.format’ 二者必需其一。json
groupGROUP“”Kafka source的消费组id。如果未指定消费组ID,则会使用自动生成的"KafkaSource-{tableIdentifier}"作为消费组ID。group_1
propertiesPROPERTIES“”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。
在这里插入图片描述

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/585388.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

<软考高项备考>《论文专题 - 36 沟通管理(4) 》

5 收尾经验教训 5.1 心得体会 1、沟通内外有别,对内可以有分歧,对外要一致。在对外的重要沟通上应安排专人负责,在沟通时应根据不同人的喜好采取不同的沟通方式。 2、在重要沟通前要制订详细的沟通计划,对要沟通的内容要有详细的…

【目标检测】yolov8结构及代码分析

yolov8代码:https://github.com/ultralytics/ultralytics yolov8的整体结构如下图(来自mmyolo): yolov8的配置文件: # Ultralytics YOLO 🚀, AGPL-3.0 license # YOLOv8 object detection model with P3-P5 outputs.…

偏好对齐RLHF-OpenAI·DeepMind·Anthropic对比分析

OpenAI paper: InstructGPT, Training language models to follow instructions with human feedback paper: Learning to summarize from human feedback Introducing ChatGPT 解密Prompt系列4介绍了InstructGPT指令微调的部分,这里只看偏好对齐的部分 样本构建…

Zookeeper-Zookeeper特性与节点数据类型详解

1.Zookeeper介绍 ZooKeeper 是一个开源的分布式协调框架,是Apache Hadoop 的一个子项目,主要用来解决分布式集群中应用系统的一致性问题。Zookeeper 的设计目标是将那些复杂目容易出错的分布式一致性服务封装起来,构成一高效可靠的原语集&…

Python列表的介绍与操作 增改查,连接,赋值,复制,清空

列表 在日常中我们通过给变量赋值来存储数据,比如 a "hello" b "world" c "你好啊" d "....."由于变量一次只能存储一个数据,但我们如果想一次存储多个数据,的话这样存储会很复杂,所以,我们可以通过列表 列表(List)是Python中的…

Linux驱动开发学习笔记6《蜂鸣器实验》

目录 一、蜂鸣器驱动原理 二、硬件原理分析 三、实验程序编写 1、 修改设备树文件 (1)添加pinctrl节点 (2)添加BEEP设备节点 (3)检查PIN 是否被其他外设使用 2、蜂鸣器驱动程序编写 3、编写测试AP…

基于js和html的骰子游戏

介绍: 1.游戏者选择“大”时,三个骰子点数之和为11-18时,游戏者获胜。2.游戏者选择“小”时,三个骰子点数之和为3-10时,游戏者获胜。3.如果游戏者选择具体点数,则根据三个骰子的点数计算,如果与…

Linux 安装Jupyter notebook 并开启远程访问

文章目录 安装Python安装pip安装Jupyter启动Jupyter Notebook1. 生成配置文件2. 创建密码3. 修改jupyter notebook的配置文件4. 启动jupyter notebook5. 远程访问jupyter notebook 安装Python 确保你的系统上已经安装了Python。大多数Linux发行版都预装了Python。你可以在终端…

【深度学习-图像分类】02 - AlexNet 论文学习与总结

论文地址:ImageNet Classification with Deep Convolutional Neural Networks 论文学习 1. 摘要 本研究训练了一个大型深度卷积神经网络(CNN),用于对ImageNet LSVRC-2010比赛中的1.2百万高分辨率图像进行分类,这些图…

SQL面试题挑战14:每年的在校人数

目录 问题:SQL解答: 问题: year表示学生入学年度,num表示对应年度录取学生人数,stu_len表示录取学生的学制;说明:例如录取年度2018学制是3年,表示该批学生在校年份为20182019、2019…

【WPF.NET开发】属性更改事件

本文内容 先决条件标识属性更改事件属性触发器 Windows Presentation Foundation (WPF) 定义几个为响应属性值的更改而引发的事件。 该属性通常是依赖项属性。 事件本身可以是路由事件,也可以是标准公共语言运行时 (CLR) 事件,具体取决于事件是应通过元…

阿里高级Java面试真题

请解释Java中的反射机制及其潜在的性能影响。 Java中的反射机制是指在运行时检查或操作类、接口、字段、方法等程序结构的能力。通过反射,你可以在运行时获取类的信息、调用类的方法、访问或修改类的字段等,而不需要在编译时就确定这些操作。反射机制为…

MYSQL 深入探索系列六 SQL执行计划

概述 好久不见了,近期一直在忙项目的事,才有时间写博客,近期频繁出现sql问题,今天正好不忙咱们看看千万级别的表到底该如何优化sql。 案例 近期有个小伙伴生产环境收到了告警,有个6千万的日志表,查询耗时大…

“undefined reference to XXX“问题总结

"undefined reference to XXX"问题总结 引言 我们在Linux下用C/C工作的时候,经常会遇到"undefined reference to XXX"的问题,直白地说就是在链接(从.cpp源代码到可执行的ELF文件,要经过预处理->编译->链接三个阶…

1panel使用指南(一)面板安装

一、1panel简介 1Panel是杭州飞致云信息科技有限公司推出的产品 [1],帮助用户实现快速建站。 [2]是一款现代化、开源的Linux服务器运维管理面板,于2023年3月推出,深度集成WordPress和Halo,一键完成域名绑定、SSL证书配置等操作&a…

draw.io学习笔记

1、链接 1.1、自动连接图形 鼠标放在图形上,点击出现的箭头,会自动出常用图形 1.2、固定连接 如果拖动其中一个图形的话,固定链接的形状会是曲线连过去。 方法:不要点击左边图形鼠标放在边框上面左边出现绿圆点鼠标左键点击图形的…

小秋SLAM入门实战ROS文章汇总

小秋SLAM入门实战教程汇总 【launch文件中如何启动gdb调试单个节点多个节点】 一个ros可执行程序可以定义几个节点? ros启动节点的launch文件你真的会写吗? 【Point Cloud ROS】用一张彩色图像和深度图像生成点云图像 【Point Cloud ROS】两张点云图像之…

Oracle研学-对象

学自B站黑马程序员 1.视图 (物化视图序列同义词oracle特有) 1.对一个SQL语句的封装,一个虚拟的表。-简化开发 视图是一种数据库对象,是从一个或者多个数据表或视图中导出的虚表,视图所对应的数据并不真正地存储在视图中&#xff…

【Linux】深度解剖环境变量

> 作者简介:დ旧言~,目前大二,现在学习Java,c,c,Python等 > 座右铭:松树千年终是朽,槿花一日自为荣。 > 目标:熟悉并掌握Linux的环境变量。 > 毒鸡汤&#x…

LeetCode刷题--- 单词搜索

个人主页:元清加油_【C】,【C语言】,【数据结构与算法】-CSDN博客 个人专栏 力扣递归算法题 http://t.csdnimg.cn/yUl2I 【C】 ​​​​​​http://t.csdnimg.cn/6AbpV 数据结构与算法 ​​​​http://t.csdnimg.cn/hKh2l 前言:这个专栏主要讲述…