使用Flink时从Kafka中读取Array[Byte]类型的Schema

使用Flink时，如果从Kafka中读取输入流，默认提供的是String类型的Schema：

val myConsumer = new FlinkKafkaConsumer08[String]("Topic名称", new SimpleStringSchema(), properties);

如果存入Kafka中的数据不是JSON，而是Protobuf类型的数据，需要用二进制的Schema进行接收，可以自己实现一个类，很简单，只有一行代码：

 class ByteArrayDeserializationSchema[T] extends AbstractDeserializationSchema[Array[Byte]]{@throws[IOException]override def deserialize(message: Array[Byte]): Array[Byte] = message
}

然后使用时，如下所示：

val myConsumer = new FlinkKafkaConsumer08[String]("Topic名称", new ByteArrayDeserializationSchema[Array[Byte]](), properties);

转载于:https://www.cnblogs.com/liugh/p/7448554.html

使用Flink时从Kafka中读取Array[Byte]类型的Schema相关推荐

【Kafka】从kafka中读取最新数据
[Kafka]从kafka中读取最新数据一.死循环无限拉取kafka数据 1.1 整体框架剖析 1.2 测试二.@KafkaListener注解实现监听kafka数据三.参考资料前情提要:我 ...
使用Spark Streaming从kafka中读取数据把数据写入到mysql 实例
文章目录一. 题目题目和数据二. pom依赖三.建表语句四. 连接kafka配置类五. 自定义分区类六. 读取数据并发送数据七. 消费数据,把数据存储到mysql 一. 题目题目和数 ...
在启动时从配置文件中读取对象
目录介绍背景使用代码如何从Web应用程序的配置文件反序列化对象介绍本技巧揭示了一种非常简单的方法,可以将Web应用程序的配置文件的各个部分作为对象读取,而无需选择模式或依赖注入. 背景在 ...
flink实时消费kafka中oracle的DML数据写入mysql
1.需要环境 zookeeper,小编安装环境为zookeeper-3.4.10 kakfa,小编安装环境为kafka_2.13-2.8.0 kafka-connect-oracle,此为kafka- ...
深度拷贝时，类中的非简单类型字段
2019独角兽企业重金招聘Python工程师标准>>> public class Test implements Cloneable {private String id;priva ...
Flink Caused by:org.apache.flink.streaming.connectors.kafka.internal.Handover$ClosedException
Flink程序从kafka中读取数据进行计算,FLink程序一启动就报以下错误,看到错误很懵逼.加班到9点没解决,第二天提前来半小时,把如下错误信息又看了一遍.具体错误如下: 错误信息1. 20/12 ...
Flink从Kafka 0.8中读取多个Topic时的问题
Flink提供了FlinkKafkaConsumer08,使用Kafka的High-level接口,从Kafka中读取指定Topic的数据,如果要从多个Topic读取数据,可以如下操作: 1.appl ...
中读取数据_Flink入门实战 (中)
一.Flink 流处理 API 1.Environment getExecutionEnvironment 创建一个执行环境,表示当前执行程序的上下文. 如果程序是独立调用的,则此方法返回本地执行环 ...
ifstream 和 ofstream 文件中读取和写入操作
导读 ofstream是从内存到硬盘,ifstream是从硬盘到内存,其实所谓的流缓冲就是内存空间. 在C++中,有一个stream这个类,所有的I/O都以这个"流"类为基础的,包 ...

使用Flink时从Kafka中读取Array[Byte]类型的Schema

使用Flink时从Kafka中读取Array[Byte]类型的Schema相关推荐

最新文章

热门文章