Ошибка сериализации Flink

Question

Я пытаюсь запустить протокол распространения Label на графике Apache Flink Gelly Graph.
Вот мой код:

        Graph<String, Long, String> ugraph = Graph.fromDataSet(vertex, edgeSet, env).getUndirected();
        DataSet<Tuple2<String, Long>> idsWithInitialLabels = DataSetUtils
            .zipWithUniqueId(graph.getVertexIds())
            .map(new MapFunction<Tuple2<Long, String>, Tuple2<String, Long>>() {
                public Tuple2<String, Long> map(Tuple2<Long, String> tuple2) throws Exception {
                    return new Tuple2<String, Long>(tuple2.f1, tuple2.f0);
                }
            }); 
        DataSet<Vertex<String, Long>> verticesWithCommunity = graph.joinWithVertices(idsWithInitialLabels,
            new VertexJoinFunction<Long, Long>() {
            public Long vertexJoin(Long vertexValue, Long inputValue) {
            return inputValue;
        }})
    .run(new LabelPropagation<String, Long, String>(10)); 

Я получил следующее сообщение об ошибке:

org.apache.flink.api.common.InvalidProgramException: объект org.apache.flink.graph.Graph$ApplyCoGroupToVertexValues@4dde0543 не сериализуемо по адресу org.apache.flink.api.java.ClosureCleaner.ensureSerializable(Clojure_Clean). org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:61) в org.apache.flink.api.java.DataSet.clean(DataSet.java:186) в org.apache.flink.api.java.operators.CoGroupOperator$CoGroupOperatorSets$CoGroupOperatorSetsPredicate$CoGroupOperatorWithoutFunction.with(CoGroupOperator.java:619) в org.apache.flink.graph.Graph.joinWithVertices(Graph.javacesstec. TextProcessor.java:405) at tu.master.ConceptDetection.TextProcessor$4.actionPerformed(TextProcessor.java:210)

Спасибо за помощь:)

Answer 1

Я предполагаю, что класс, содержащий код вашего графика, не Serializable. Анонимные классы в Java на самом деле являются нестатическими внутренними классами, что означает, что они имеют ссылку на содержащий классthis(см. этот ответ). Если содержащий класс неSerializable, то this ссылка не будет сериализована, как и анонимный класс.

Это объясняет, почему переход на лямбда-выражение приводит к его сериализации. Лямбда-выражения не являются анонимными классами, поэтому они не захватывают автоматически неявныйthis ссылка.

Что не объясняет, так это почему MapFunctionкак анонимный класс все еще работает. Если у вас все еще есть этот код, @Nesrine, мне было бы любопытно, как выглядит весь класс.