ApacheFlink在序列化POJO类时出现KryoException错误
创始人
2024-09-05 20:00:41
0次

出现此错误的原因是因为Kryo序列化在处理POJO类时遇到了无法序列化的字段或方法。解决此问题,需要在POJO类中使用注释将无需序列化的字段或方法排除在序列化范围之外。

以下是一个示例,演示如何在POJO类中使用注释来处理此问题:

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.api.java.functions.FunctionAnnotation.ForwardedFields;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.typeutils.runtime.kryo.Serializers;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;

import java.util.Properties;

public class MyApp {

    public static void main(String[] args) throws Exception {

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "localhost:9092");
        props.setProperty("group.id", "test");

        DataStream stream = env
                .addSource(new FlinkKafkaConsumer<>("my-topic", new CustomDeserializationSchema(), props));

        DataStream outputStream = stream
                .map(new CustomMapFunction())
                .keyBy(tuple -> tuple.f0)
                .addSink(new FlinkKafkaProducer<>("my-other-topic", new CustomSerializationSchema("my-other-topic"), props));

        env.execute("MyApp");
    }

    private static class CustomDeserializationSchema implements DeserializationSchema {
        private static final long serialVersionUID = 1L;
        private final ObjectMapper objectMapper = new ObjectMapper();
        @Override
        public String deserialize(byte[] message) throws IOException {
            return objectMapper.readValue(message, String.class);
        }
    }

    private static class CustomMapFunction

相关内容

热门资讯

热门推荐!wepoker俱乐部... 热门推荐!wepoker俱乐部辅助,微扑克透视是真的吗工具(确实有开挂)AI 是一款可以让一直输的玩...
传递经验!wpk作弊最怕三个东... 传递经验!wpk作弊最怕三个东西,WPK破解软件脚本(一直有辅助器)AI 是一款可以让一直输的玩家修...
必看攻略!德扑之心免费透视,德... 一款 AI 游戏攻略黑科技工具上线,查攻略效率直接开挂。它整合版本更新、装备搭配、BOSS 打法、牌...
每日必看!wepoker辅助是... 每日必看!wepoker辅助是真的假的,wepoker开挂透视教程下载(起初真的是有开挂)AI 是一...
玩家必看科普!wpk透视辅助靠... 一款 AI 游戏攻略黑科技工具上线,查攻略效率直接开挂。它整合版本更新、装备搭配、BOSS 打法、牌...
可靠教程!德普之星辅助器app... 很多人使用云端 AI 时,都担心德普之星辅助器app、内部资料上传带来隐私风险。这款本地运行的 AI...
教程攻略!德普之星辅助工具如何... 您好:德普之星辅助工具如何设置这款游戏可以开挂的,确实是有挂的,很多玩家在这款游戏中打牌都会发现很多...
9分钟了解!WPK作弊,微扑克... WPK作弊AI 游戏攻略黑科技上线本地缓存功能,离线查阅能力开挂。提前下载对应游戏攻略数据包,断网状...
科普攻略!wepoker透视脚... 『标题:当前标题』主打赛后复盘的 AI 游戏攻略黑科技走红,对局总结能力开挂。上传对局回放或者战绩截...
一秒答解!wepoker辅助透... 您好,一秒答解!wepoker辅助透视软件,德扑之星是如何开挂的插件(其实真的有透视)这款游戏可以开...