序列化flink中kafka的对象

Question

我试图使用flink从kafka中读取数据，执行一些函数，并将结果返回到不同的kafka主题，但出现以下错误`组织。阿帕奇。Flink。应用程序编程接口。常见的InvalidProgrameException:MapFunction的实现不可序列化。对象可能包含或引用不可序列化的字段。

我收到了来自kafka的消息-对其进行了一些操作，并返回了一个对象列表，我想发送到不同的主题。

class Wrapper implements Serializable{
        @JsonProperty("viewBuilderRequests")
        private ArrayList<ViewBuilderRequest> viewBuilderRequests;

        public Wrapper(){}

        public Wrapper(ArrayList<ViewBuilderRequest> viewBuilderRequests) {
            this.viewBuilderRequests = viewBuilderRequests;
        }

        public List<ViewBuilderRequest> getViewBuilderRequests() {
            return viewBuilderRequests;
        }

        public void setViewBuilderRequests(ArrayList<ViewBuilderRequest> viewBuilderRequests) {
            this.viewBuilderRequests = viewBuilderRequests;
        }
    }



public class ViewBuilderRequest implements Serializable {
    private CdmId cdmId
    private ViewBuilderOperation operation
    private List<ViewUserSystemIdentifier> viewUserSystemIdentifiers
    public ViewBuilderRequest(){
}

    public CdmId getCdmId() {
        return cdmId;
    }

    public void setCdmId(CdmId cdmId) {
        this.cdmId = cdmId;
    }

    public ViewBuilderOperation getOperation() {
        return operation;
    }

    public void setOperation(ViewBuilderOperation operation) {
        this.operation = operation;
    }

    public List<ViewUserSystemIdentifier> getViewUserSystemIdentifiers() {
        return viewUserSystemIdentifiers;
    }

    public void setViewUserSystemIdentifiers(List<ViewUserSystemIdentifier> viewUserSystemIdentifiers) {
        this.viewUserSystemIdentifiers = viewUserSystemIdentifiers;
    }

    public enum ViewBuilderOperation implements Serializable{
        Create, Update,Delete
    }




private MapFunction<String, Wrapper> parseAndSendToGraphProcessing = s ->{
    UserMatchingRequest userMatchingRequest = objectMapper.readValue(s, UserMatchingRequest.class);
    Wrapper wrapper = new Wrapper(janusGraphDataProcessing.handleMessage(userMatchingRequest));
    return wrapper;
};

内部类也实现了可序列化

此代码引发异常：

dataStream。映射（parseAndSendToGraphProcessing）。addSink（新Flinkkapkaproducer

我还对这两个对象进行了反序列化。

public class WrapperSchema implements DeserializationSchema<Wrapper>, SerializationSchema<Wrapper> {
//        private final static ObjectMapper objectMapper = new ObjectMapper().configure(MapperFeature.ACCEPT_CASE_INSENSITIVE_PROPERTIES, true);

    static ObjectMapper objectMapper = new ObjectMapper();

    @Override
        public Wrapper deserialize(byte[] message) throws IOException {
            return objectMapper.readValue(message, Wrapper.class);
        }

        @Override
        public boolean isEndOfStream(Wrapper nextElement) {
            return false;
        }

        @Override
        public byte[] serialize(Wrapper element) {
//            return element.toString().getBytes();
            if(objectMapper == null) {
                objectMapper.setVisibility(PropertyAccessor.FIELD, JsonAutoDetect.Visibility.ANY);
                objectMapper = new ObjectMapper();
            }
            try {
                String json = objectMapper.writeValueAsString(element);
                return json.getBytes();
            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }

            return new byte[0];
        }

        @Override
        public TypeInformation<Wrapper> getProducedType() {
            return TypeInformation.of(Wrapper.class);
        }
}




Answer 1

要使flink工作，您的消息和地图函数都必须是可序列化的。

据我所知，你的消息看起来是可序列化的。

但是你的map函数不是。有时候很难让lambda序列化。我认为在您的案例中，问题在于parseAndSendToGraphProcessing使用的是objectMapper和janusGraphDataProcessing，它们必须是可序列化的。

我的猜测是janusGraphDataProcessing是不可序列化的（如果您使用的是jackson 2.1或更高版本，则OjbectMapper是）。

如果是这种情况，那么一种解决方法是编写一个自定义RichMapFunction类，该类将把janusGraphDataProcessing存储为瞬态变量，并在其open函数中对其进行初始化。

private MapFunction<String, Wrapper> parseAndSendToGraphProcessing = s ->{
    UserMatchingRequest userMatchingRequest = objectMapper.readValue(s, UserMatchingRequest.class);
    Wrapper wrapper = new Wrapper(janusGraphDataProcessing.handleMessage(userMatchingRequest));
    return wrapper;
};