如何用自定义对象处理和聚合kafka流？

hjqgdpho  于 2021-06-06  发布在  Kafka
关注(0)|答案(1)|浏览(461)

所以基本上我有会计类。我有数据。我想把这些东西放到我和制作人的主题里。现在还可以。稍后，我想用kafka流进行聚合，但是我不能，因为我的配置中有些serde属性是错误的，我想：/。我不知道错误在哪里。我的制作人工作得很好，但我不能汇总。有人帮我看看Kafka的密码吗？我的帐户类别：

public class Account {

    private long fromId;
    private long amount;
    private long toId;
    private ZonedDateTime time;
}

有两个类序列化和反序列化为我的帐户类。序列化程序：

public class AccountSerializer implements Serializer {

    private static final Charset CHARSET = Charset.forName("UTF-8");
    static private Gson gson = new Gson();

    @Override
    public void configure(Map map, boolean b) {

    }

    @Override
    public byte[] serialize(String s, Object o) {
        String line = gson.toJson(o);
        // Return the bytes from the String 'line'
        return line.getBytes(CHARSET);
    }

    @Override
    public void close() {

    }
}

反序列化程序：

public class AccountDeserializer implements Deserializer {
    private static final Charset CHARSET = Charset.forName("UTF-8");
    static private Gson gson;

    static {
        gson = new Gson();
    }

    @Override
    public void configure(Map map, boolean b) {

    }

    @Override
    public Object deserialize(String s, byte[] bytes) {
        try {
            // Transform the bytes to String
            String person = new String(bytes, CHARSET);
            // Return the Person object created from the String 'person'
            return gson.fromJson(person, Account.class);
        } catch (Exception e) {
            throw new IllegalArgumentException("Error reading bytes! Yanlış", e);
        }
    }

    @Override
    public void close() {

    }
}

我为Kafka的accountserde课程：

public class AccountSerde implements Serde<Object> {

    private AccountSerializer accountSerializer;
    private AccountDeserializer accountDeserializer;

    @Override
    public void configure(Map<String, ?> map, boolean b) {

    }

    @Override
    public void close() {
        accountSerializer.close();
        accountDeserializer.close();

    }

    @Override
    public Serializer<Object> serializer() {
        return accountSerializer;
    }

    @Override
    public Deserializer<Object> deserializer() {
        return accountDeserializer;
    }
}

我的Kafka制作人：

public static void main(String[] args) {

        DataAccess dataAccess = new DataAccess();
        List<Account> accountList = dataAccess.read();

        final Logger logger = LoggerFactory.getLogger(Producer.class);
        Properties properties = new Properties();

        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092");
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,LongSerializer.class.getName());
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,AccountSerializer.class.getName());

        KafkaProducer<Long,Account> producer = new KafkaProducer<>(properties);

        for (Account account : accountList) {

            ProducerRecord<Long,Account> record = new ProducerRecord<Long, Account>("bank_account",account.getFromId(),account);

            producer.send(record, new Callback() {
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (e == null) {

            logger.info("Record sent successfully. \n "+ "Topic : "+recordMetadata.topic() +"\n"+
                            "Partition : " + recordMetadata.partition() + "\n"+
                            "Offset : " +recordMetadata.offset() +"\n"+
                            "Timestamp: " +recordMetadata.timestamp() +"\n");
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e1) {
                        e1.printStackTrace();
                    }

                }
                else{
                    logger.info("Error sending producer");
                }
            }
        });
    }

    producer.flush();
    producer.close();
}

这是我想尝试聚合的类，我的kafka流类。

public static void main(String[] args) {
        System.out.println();

        Properties properties = new Properties();
        properties.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.01:9092");
        properties.setProperty(StreamsConfig.APPLICATION_ID_CONFIG,"demo-kafka-streams");
        properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,AccountDeserializer.class.getName());
        properties.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.LongSerde);
        properties.setProperty(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, AccountSerde.class.getName());
        //create a topology

        StreamsBuilder streamsBuilder = new StreamsBuilder();

        KStream<Long, Account> inputTopic = streamsBuilder.stream("bank_account");

        KTable<Long, Long> aggregate = inputTopic.groupByKey().aggregate(
                () -> 0L,
                (key, current, oldBalance) -> current.getAmount() + oldBalance);

        aggregate.toStream().to("son");

        KafkaStreams streams = new KafkaStreams(streamsBuilder.build(),properties);

        streams.start();

        System.out.println(streams.toString());

        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));

    }

我试过了，我的制作人工作得很好，并且发送对象。但是由于错误，我无法尝试我的聚合代码是否正常工作。它给了我

[demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89-StreamThread-1] ERROR org.apache.kafka.streams.errors.LogAndFailExceptionHandler - Exception caught during Deserialization, taskId: 0_0, topic: bank_account, partition: 0, offset: 0
java.lang.NullPointerException
    at org.apache.kafka.streams.processor.internals.SourceNode.deserializeValue(SourceNode.java:63)
    at org.apache.kafka.streams.processor.internals.RecordDeserializer.deserialize(RecordDeserializer.java:66)
    at org.apache.kafka.streams.processor.internals.RecordQueue.addRawRecords(RecordQueue.java:97)
    at org.apache.kafka.streams.processor.internals.PartitionGroup.addRawRecords(PartitionGroup.java:117)
    at org.apache.kafka.streams.processor.internals.StreamTask.addRecords(StreamTask.java:638)
    at org.apache.kafka.streams.processor.internals.StreamThread.addRecordsToTasks(StreamThread.java:936)
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:831)
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:767)
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:736)
[demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89-StreamThread-1] INFO org.apache.kafka.streams.processor.internals.StreamThread - stream-thread [demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89-StreamThread-1] State transition from RUNNING to PENDING_SHUTDOWN
[demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89-StreamThread-1] INFO org.apache.kafka.streams.processor.internals.StreamThread - stream-thread [demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89-StreamThread-1] Shutting down
[demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89-StreamThread-1] INFO org.apache.kafka.clients.producer.KafkaProducer - [Producer clientId=demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89-StreamThread-1-producer] Closing the Kafka producer with timeoutMillis = 9223372036854775807 ms.
[demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89-StreamThread-1] INFO org.apache.kafka.streams.processor.internals.StreamThread - stream-thread [demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89-StreamThread-1] State transition from PENDING_SHUTDOWN to DEAD
[demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89-StreamThread-1] INFO org.apache.kafka.streams.KafkaStreams - stream-client [demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89] State transition from RUNNING to ERROR
[demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89-StreamThread-1] WARN org.apache.kafka.streams.KafkaStreams - stream-client [demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89] All stream threads have died. The instance will be in error state and should be closed.
[demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89-StreamThread-1] INFO org.apache.kafka.streams.processor.internals.StreamThread - stream-thread [demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89-StreamThread-1] Shutdown complete
Exception in thread "demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89-StreamThread-1" org.apache.kafka.streams.errors.StreamsException: Deserialization exception handler is set to fail upon a deserialization error. If you would rather have the streaming pipeline continue after a deserialization error, please set the default.deserialization.exception.handler appropriately.
    at org.apache.kafka.streams.processor.internals.RecordDeserializer.deserialize(RecordDeserializer.java:80)
    at org.apache.kafka.streams.processor.internals.RecordQueue.addRawRecords(RecordQueue.java:97)
    at org.apache.kafka.streams.processor.internals.PartitionGroup.addRawRecords(PartitionGroup.java:117)
    at org.apache.kafka.streams.processor.internals.StreamTask.addRecords(StreamTask.java:638)
    at org.apache.kafka.streams.processor.internals.StreamThread.addRecordsToTasks(StreamThread.java:936)
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:831)
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:767)
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:736)
Caused by: java.lang.NullPointerException
    at org.apache.kafka.streams.processor.internals.SourceNode.deserializeValue(SourceNode.java:63)
    at org.apache.kafka.streams.processor.internals.RecordDeserializer.deserialize(RecordDeserializer.java:66)
    ... 7 more
Javaapache-kafkaapache-kafka-streamsserialization

1条答案

按热度按时间
2hh7jdfx

2hh7jdfx1#

你从不初始化字段，所以你会得到一个npe
您还应该将serde类型更改为实际的类

public class AccountSerde implements Serde<Account> {

    // These are both null unless you initialize them 
    private AccountSerializer accountSerializer;
    private AccountDeserializer accountDeserializer;

此外，还需要根据该值修复ip地址，该值不是有效的ip地址

"127.0.01:9092"
赞(0）分享  回复(0）举报  2021-06-07
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备2022004421号-2 NULL123 https://www.null123.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com