正文 

当前位置： 首页 > IT > java教程

flinkkafka消费pojo类型数据实战详解(代码片段)

bitcarmanlee bitcarmanlee     2022-12-10     373

关键词：

1.pojo类数据接口

在实际生产环境中，我们一般会将数据封装成一个pojo类(或者其他rpc框架通过IDL生成一个java类)，这样能方便我们后续的数据传输与解析。该pojo类就相当于标准数据接口，可以在任何地方被引用或者使用。下面我们来看看，怎么通过kafka的producer来生产这些pojo类数据，又怎么通过flink或者kafka的consumer来消费这些数据。

2.kafka producer生产数据

2.1 定义pojo类

首先，我们定义一个pojo类

public class User 
    public String name;
    public int age;

    public User() 

    public User(String name, int age) 
        this.name = name;
        this.age = age;
    

    public String getName() 
        return name;
    

    public void setName(String name) 
        this.name = name;
    

    public int getAge() 
        return age;
    

    public void setAge(int age) 
        this.age = age;
    

    @Override
    public String toString() 
        return "user" +
                "name='" + name + '\\'' +
                ", age=" + age +
                "";

我们定义了一个User类，该类有两个字段name与age。需要注意的是，该类必须要包含有默认的构造函数，否则后续代码使用过程中会出现问题，具体可以参考如下链接

https://stackoverflow.com/questions/7625783/jsonmappingexception-no-suitable-constructor-found-for-type-simple-type-class

2.2 定义序列化类

import org.apache.kafka.common.serialization.Serializer;
import org.codehaus.jackson.map.ObjectMapper;

import java.io.IOException;
import java.util.Map;

/**
 * author: wanglei
 * create: 2022-09-26
 */
public class UserSerializer implements Serializer<User> 

    @Override
    public void configure(Map configs, boolean isKey) 
    

    @Override
    public byte[] serialize(String topic, User user) 
        if (user == null) return null;
        ObjectMapper objectMapper = new ObjectMapper();
        try 
            return objectMapper.writeValueAsString(user).getBytes();
         catch (IOException e) 
            e.printStackTrace();
        
        return null;
    

    @Override
    public void close()

上面类的作用，是将User对象序列化的过程，方便后面的数据传输。上面使用了ObjectMapper类进行序列化，需要引入如下依赖

        <dependency>
            <groupId>org.codehaus.jackson</groupId>
            <artifactId>jackson-mapper-asl</artifactId>
            <version>1.9.13</version>
        </dependency>

2.3 实现producer

import edu.bit.leilei.serialize.User;
import edu.bit.leilei.serialize.UserSerializer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * author: wanglei
 * create: 2022-09-26
 */
public class PojoProducer 

    public static void main(String[] args) 
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, UserSerializer.class.getName());
        KafkaProducer<String, User> producer = new KafkaProducer<String, User>(properties);

        String topic = "pojotest";
        for(int i=0; i<5; i++) 
            User user = new User("my name-" + i, i);
            ProducerRecord<String, User> record = new ProducerRecord<String, User>(topic, "key-"+i, user);
            producer.send(record);
        

        producer.close();

上面的代码，生成了5个User对象。将代码先运行2次，总共生成了10个user对象。再将里面的一行稍作修改

            User user = new User("my name-" + i, i*10);

再运行一次，这样topic里面总共就有了15个对象。

3.kafka consumer消费数据

3.1 编写反序列化类

consumer要消费数据，首先需要做的是对数据进行反序列化。我们先来编写反序列化类代码

import org.apache.kafka.common.serialization.Deserializer;
import org.codehaus.jackson.map.ObjectMapper;

import java.io.IOException;
import java.util.Map;

/**
 * author: wanglei
 * create: 2022-09-26
 */
public class UserDeserializer implements Deserializer<User> 
    @Override
    public void configure(Map<String, ?> configs, boolean isKey) 
    

    @Override
    public User deserialize(String topic, byte[] data) 
        ObjectMapper objectMapper = new ObjectMapper();
        try 
            return objectMapper.readValue(data, User.class);
         catch (IOException e) 
            e.printStackTrace();
        
        return null;
    

    @Override
    public void close()

上面同样用到的是ObjectMapper对byte[]数据进行反序列化。

3.2 kafka consumer代码

import edu.bit.leilei.serialize.User;
import edu.bit.leilei.serialize.UserDeserializer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Arrays;
import java.util.Properties;

/**
 * author: wanglei
 * create: 2022-09-26
 */
public class PojoConsumer 
    public static void main(String[] args) 
        String topic = "pojotest";
        String groupId = "group_leilei";

        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, UserDeserializer.class.getName());
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("enable.auto.commit", "true");//设置为自动提交
        props.put("auto.offset.reset", "earliest");

        KafkaConsumer<String, User> consumer = new KafkaConsumer<String, User>(props);
        consumer.subscribe(Arrays.asList(topic));

        while(true) 
            ConsumerRecords<String, User> records = consumer.poll(1L);
            for(ConsumerRecord record : records) 
                System.out.printf("patition = %d , offset = %d, key = %s, value = %s%n",
                        record.partition(), record.offset(), record.key(), record.value());

上面的代码中，指定了反序列化类为UserDeserializer，并且从topic的最早位置开始进行消费。

最后输出的结果为

patition = 0 , offset = 0, key = key-0, value = username='my name-0', age=0
patition = 0 , offset = 1, key = key-1, value = username='my name-1', age=1
patition = 0 , offset = 2, key = key-2, value = username='my name-2', age=2
patition = 0 , offset = 3, key = key-3, value = username='my name-3', age=3
patition = 0 , offset = 4, key = key-4, value = username='my name-4', age=4
patition = 0 , offset = 5, key = key-0, value = username='my name-0', age=0
patition = 0 , offset = 6, key = key-1, value = username='my name-1', age=1
patition = 0 , offset = 7, key = key-2, value = username='my name-2', age=2
patition = 0 , offset = 8, key = key-3, value = username='my name-3', age=3
patition = 0 , offset = 9, key = key-4, value = username='my name-4', age=4
patition = 0 , offset = 10, key = key-0, value = username='my name-0', age=0
patition = 0 , offset = 11, key = key-1, value = username='my name-1', age=10
patition = 0 , offset = 12, key = key-2, value = username='my name-2', age=20
patition = 0 , offset = 13, key = key-3, value = username='my name-3', age=30
patition = 0 , offset = 14, key = key-4, value = username='my name-4', age=40

4.flink消费pojo类型数据

我们尝试使用flink来消费pojo类的数据。

4.1 反序列化类

同样的，首先也需要编写flink中反序列化的schema类。

import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.codehaus.jackson.map.ObjectMapper;

import java.io.IOException;

/**
 * author: wanglei
 * create: 2022-09-27
 */
public class UserFlinkDeserializer implements DeserializationSchema<User> 
    @Override
    public User deserialize(byte[] message) throws IOException 
        ObjectMapper objectMapper = new ObjectMapper();
        try 
            return objectMapper.readValue(message, User.class);
         catch (IOException e) 
            e.printStackTrace();
        
        return null;
    

    @Override
    public boolean isEndOfStream(User nextElement) 
        return false;
    

    @Override
    public TypeInformation<User> getProducedType() 
        return TypeInformation.of(User.class);

2.编写flink相关代码

import edu.bit.leilei.serialize.User;
import edu.bit.leilei.serialize.UserFlinkDeserializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09;

import java.util.Properties;

/**
 * author: wanglei
 * create: 2022-09-27
 */
public class StreamPojo 

    public static void main(String[] args) throws Exception 
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        String topic = "pojotest";
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.put("group.id", topic);

        FlinkKafkaConsumer09<User> myConsumer = new FlinkKafkaConsumer09<User>(topic, new UserFlinkDeserializer(), properties);
        myConsumer.setStartFromEarliest();
        DataStream<User> stream = env.addSource(myConsumer);

        stream.print();
        env.execute();

代码运行以后得到的输出结果为

username='my name-0', age=0
username='my name-1', age=1
username='my name-2', age=2
username='my name-3', age=3
username='my name-4', age=4
username='my name-0', age=0
username='my name-1', age=1
username='my name-2', age=2
username='my name-3', age=3
username='my name-4', age=4
username='my name-0', age=0
username='my name-1', age=10
username='my name-2', age=20
username='my name-3', age=30
username='my name-4', age=40

第一篇数据分析项目实战:用户消费行为分析

...A本篇文章以模仿为主,利用pandas进行数据处理，分析用户消费行为。数据来源CDNow网站的用户购买明细。一共有用户ID，购买日期，购买数量，购买金额四个字段。分析步骤第一部分：数据类型的处理—字段的清洗缺失值的处理、... 查看详情

kotlin入门到精通（实战）基本类型和访问修饰符详解

<<<返回总目录本章目录Kotlin数据类型1.如果变量可以为null2.如果作为泛型参数3.如何声明基本类型数组4.Kotlin数据类型为什么不共用Java那一套Kotlin访问修饰符1.internal修饰符2.protected修饰符在Kotlin和Java中的异同点1)protected修... 查看详情

redis实战|5种redis数据类型详解(代码片段)

...其实还有另外一个优势，就是Redis提供了更加丰富的数据类型，这使得Redis有着更加广泛的使用场景。那Redis提供给用户的有哪些数据类型呢？主要有：string（字符串）、List（列表）、Set（集合）、Hash（哈希）、Zset（有序集合）... 查看详情

flinkflink消费kafka消费组死掉markingthecoordinatordeadforgroup造成数据重复消费

...提交呢？2.3监控日志1.概述首先参考几个案例：【Flink】FlinkKafka消费卡死消费组卡死topic无写入实际有数据topic正常Kafka:kafka无法消费的情况在博客【kafka】消费组死掉kafkaMarkingthecoordinatordeadforgroup中消费组死掉，但是不会造成数据... 查看详情

项目实战--用户消费数据分析(代码片段)

...引入包一、数据预处理二、按月对数据分析三、用户个体消费数据分析四、用户消费行为分析五、用户的生命周期引入包本项目所用数据为【密码：pfj6】:CDNOW_master.txtimportnumpyasnpimportpandasaspdfrompandasimportDataFrame,Seriesimportmatplot... 查看详情

yaml详解与实战(代码片段)

...时间阅读最新技术文章。文章目录YAML简介基本语法数据类型标量对象数组文本块显示指定类型引用单文件多配置YAML简介YAML，即YAMLAin’taMarkupLanguage（YAML不是一种标记语言）的递归缩写。YAML其实意思是Ye 查看详情

我在数据库中有 tinyBlob 数据类型，写 pojo 时对应的数据类型应该是啥？

】我在数据库中有tinyBlob数据类型，写pojo时对应的数据类型应该是啥？【英文标题】：IhavetinyBlobdatatypeindatabse,whatshouldbecorrespodingdatatypewhilewritingpojo?我在数据库中有tinyBlob数据类型，写pojo时对应的数据类型应该是什么？【发布时... 查看详情

mybatis的mapper.xml文件详解

...    #表示一个占位符号，#接收输入参数，类型可以是简单类型，pojo、hashmap。    如果接收简单类型，#中可以写成value或其它名称。    #接收pojo对象值，通过OGNL读取对象中的属性值，... 查看详情

kafka实战-消费者offset重置问题(代码片段)

kafka实战-消费者offset重置问题背景问题现象分析原因问题解决附-常见的消费者配置描述和调优方案1.max.poll.records2.fetch.max.bytes3.heartbeat.interval.ms4.max.partition.fetch.bytes5.fetch.max.bytes6.session.timeout.ms7.auto.offset.reset8.max.poll.interval.ms9.max.p... 查看详情

do,dto,vo,pojo详解

...回复「Java开发手册」获得。　　•DO（DataObject）：与数据库表结构一一对应，通过DAO层向上传输数据源对象。　 查看详情

mybatis传入参数类型parametertype和输出结果类型resulttype详解

...性，parameterType属性用于对应的mapper接口方法接受的参数类型，resultType用于指定sql输出的结果类型。resultType：指定sql输出结果类型，总共就两种：1.基本数据类型。2.pojo类类型。mybatis将 查看详情

sparksql数据加载和保存实战

...存到文件或者说与具体的格式来指明我们要读取的文件的类型以及与具体的格式来指出我们要输出的文件是什么类型。 二：SparkSQL读写数据代码实战：6 查看详情

sparksql数据加载和保存实战

...存到文件或者说与具体的格式来指明我们要读取的文件的类型以及与具体的格式来指出我们要输出的文件是什么类型。 二：SparkSQL读写数据代码实战：i 查看详情

kafkakafka查看groupcoordinator

...kafkaGroupdead消费组死掉以及GroupCoordinatorRequest使用【Flink】FlinkKafka消费卡死消费组卡死topic无写入实际有数据topic正常 查看详情

springmvc详解(学习总结)(代码片段)

...5.5JSP页面的转发和重定向六、SpringMVC数据绑定6.1基本数据类型6.2包装类6.3数组类型6.4POJO(java对象)6.5List6.6JSON七、SpringMVC视图层解析7.1业务数据绑定到request域对象7.1.1Map7.1.2Model7.1.3ModelAndView7.1.4Servlet的API7.1.5@ModelAttribute7.2业务数... 查看详情

一文详解：springbootkafka从头消费

系统介绍整个系统可以从功能上分为3块：业务系统：在上游有很多的业务系统，业务系统的运行产生很多的数据，这些数据分散在很多的数据库中，大部分是MySQL数据库数据智能平台：数据智能平台属于中... 查看详情

r语言tidyr包gather()函数实战详解：数据收缩从宽表到窄表

R语言tidyr包gather()函数实战详解：数据收缩、从宽表到窄表目录R语言tidyr包gather()函数实战详解：数据收缩、从宽表到窄表收缩两列数据 查看详情

r语言tidyr包unite()函数实战详解：多个数据列合并为一列

R语言tidyr包Unite()函数实战详解：多个数据列合并为一列目录R语言tidyr包Unite()函数实战详解：多个数据列合并为一列 查看详情