Flink kafka 数据转成自己需要的实体类

1，主代码：

package application.storm;import entry.ConsumerDeserializationSchema;
import entry.TopicInfoTest;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
import org.slf4j.LoggerFactory;import java.util.Properties;public class FlinkTest {private static org.slf4j.Logger logger = LoggerFactory.getLogger(FlinkTest.class);public static void main(String[] args) throws Exception {final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();args = new String[]{"--input-topic", "dianyou_wxgz3", "--bootstrap.servers", "node2.hadoop:9091,node3.hadoop:9091,node2.hadoop:9092,node3.hadoop:9092","--zookeeper.connect", "node1.hadoop:2181,node2.hadoop:2181,node3.hadoop:2181", "--group.id", "cc1"};logger.error("************进入了程序**************");ParameterTool parameterTool = ParameterTool.fromArgs(args);env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);Properties pros = parameterTool.getProperties();//todo 指定输入数据为kafka topicDataStream<TopicInfoTest> kafkaDstream_test = env.addSource(new FlinkKafkaConsumer010<TopicInfoTest>("dianyou_wxgz3",new ConsumerDeserializationSchema(TopicInfoTest.class),
//                new MySchema(),pros).setStartFromGroupOffsets()).setParallelism(4);kafkaDstream_test.print();env.execute("startExecute");}}

2，实体类与自定义格式

package entry;import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.TypeExtractor;import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.CharBuffer;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;public class ConsumerDeserializationSchema implements DeserializationSchema<TopicInfoTest> {private Class<TopicInfoTest> clazz;public ConsumerDeserializationSchema(Class<entry.TopicInfoTest> clazz) {this.clazz = clazz;}@Overridepublic TopicInfoTest deserialize(byte[] message) throws IOException {ByteBuffer buffer = ByteBuffer.wrap(message).order(ByteOrder.LITTLE_ENDIAN);String mess = byteBuffertoString(buffer);//封装为POJO类System.out.println("mess = " + mess);return new TopicInfoTest(mess);}@Overridepublic boolean isEndOfStream(TopicInfoTest t) {return false;}@Overridepublic TypeInformation<TopicInfoTest> getProducedType() {return TypeExtractor.getForClass(clazz);}public static String byteBuffertoString(ByteBuffer buffer) {Charset charset = null;CharsetDecoder decoder = null;CharBuffer charBuffer = null;try {charset = Charset.forName("UTF-8");decoder = charset.newDecoder();// charBuffer = decoder.decode(buffer);//用这个的话，只能输出来一次结果，第二次显示为空charBuffer = decoder.decode(buffer.asReadOnlyBuffer());return charBuffer.toString();} catch (Exception ex) {ex.printStackTrace();return "";}}
}

package entry;import java.io.Serializable;public class TopicInfoTest implements Serializable {private String topicName;public String getTopicName() {return topicName;}public void setTopicName(String topicName) {this.topicName = topicName;}public TopicInfoTest(String topicName) {this.topicName = topicName;}@Overridepublic String toString() {return "TopicInfoTest{" +"topicName='" + topicName + '\'' +'}';}
}

3，如果写成泛型：

package entry;import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.TypeExtractor;import java.io.IOException;public class ConsumerDeserializationSchema<T> implements DeserializationSchema<T> {private Class<T> clazz;public ConsumerDeserializationSchema(Class<T> clazz) {this.clazz = clazz;}@Overridepublic T deserialize(byte[] bytes) throws IOException {//确保 new String(bytes) 是json 格式，如果不是，请自行解析System.out.println("bytes = " + new String(bytes));System.out.println("json-parser"+JSON.parseObject(new String(bytes), clazz));return JSON.parseObject(new String(bytes), clazz);}@Overridepublic boolean isEndOfStream(T t) {return false;}@Overridepublic TypeInformation<T> getProducedType() {return TypeExtractor.getForClass(clazz);}
}

测试数据为：

{"topicName":"1"}

打印结果：

Flink kafka 数据转成自己需要的实体类相关推荐

Storm 消费Kafka数据及相关异常解决
Storm 消费Kafka数据及相关异常解决问题 KafkaTopoDemo类 bolt类问题 storm报错:Exception in thread "main" java. ...
(一)EasyExcel的使用（读取数据到实体类即绑定实体类）
最近遇到了一个excel简单的导入导出的需求,因此就对easyexcel第三方插件的使用做一点总结,大家可以看一看,可能会对你有点帮助. 目录前言: 1.引入easyexcel相关依赖 2.创建对应 ...
2017-4-20实体类,数据访问类.字符串攻击.防攻击
程序分三层:界面层.业务逻辑层.数据访问层这里主要操作的是数据访问层,数据访问层又分为: 1.实体类 2.数据访问类存放位置:APP_Code文件中一.实体类最简单的封装把数据库的表名变成类 ...
《从0到1学习Flink》—— Flink 读取 Kafka 数据批量写入到 MySQL
 前言之前其实在 <从0到1学习Flink>-- 如何自定义 Data Sink ? 文章中其实已经写了点将数据写入到 MySQL,但是一些配置化的 ...
kafka学习_《从0到1学习Flink》—— Flink 读取 Kafka 数据写入到 RabbitMQ
前言之前有文章 <从0到1学习Flink>-- Flink 写入数据到 Kafka 写过 Flink 将处理后的数据后发到 Kafka 消息队列中去,当然我们常用的消息队列可不止这一种, ...
实时数仓大数据 Hadoop flink kafka
⼀.实时数仓建设背景实时需求⽇趋迫切⽬前各⼤公司的产品需求和内部决策对于数据实时性的要求越来越迫切,需要实时数仓的能⼒来赋能.传统离线数仓的数据时效性是 T+1,调度频率以天为单位,⽆法⽀撑实时 ...
写入mysql_《从0到1学习Flink》—— Flink 读取 Kafka 数据批量写入到 MySQL
前言之前其实在 <从0到1学习Flink>-- 如何自定义 Data Sink ? 文章中其实已经写了点将数据写入到 MySQL,但是一些配置化的东西当时是写死的,不能够通用,最近知识星 ...
flink 写入到es_《从0到1学习Flink》—— Flink 写入数据到 Kafka
前言之前文章 <从0到1学习Flink>-- Flink 写入数据到 ElasticSearch 写了如何将 Kafka 中的数据存储到 ElasticSearch 中,里面其实就已经用 ...
flink读不到kafka数据问题
[1]搭建集群 [2]测试kafka集群没问题后, 根据[3]中开始用flink消费kafka数据,发现啥也么有调试办法如下: ①去$KAFKA/logs下面看了下,是有相关主题的文件夹生成的. ② ...

Flink kafka 数据转成自己需要的实体类

Flink kafka 数据转成自己需要的实体类相关推荐

最新文章

热门文章