10

在使用带有以下代码的 Apache Flink 时：

DataStream<List<String>> result = source.window(Time.of(1, TimeUnit.SECONDS)).mapWindow(new WindowMapFunction<String, List<String>>() {

    @Override
    public void mapWindow(Iterable<String> iterable, Collector<List<String>> collector) throws Exception {
        List<String> top5 = Ordering.natural().greatestOf(iterable, 5);
        collector.collect(top5);
    }
}).flatten();

我得到了这个例外

Caused by: java.lang.UnsupportedOperationException
    at java.util.Collections$UnmodifiableCollection.add(Collections.java:1055)
    at com.esotericsoftware.kryo.serializers.CollectionSerializer.read(CollectionSerializer.java:116)
    at com.esotericsoftware.kryo.serializers.CollectionSerializer.read(CollectionSerializer.java:22)
    at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:761)
    at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.deserialize(KryoSerializer.java:211)
    at org.apache.flink.streaming.runtime.streamrecord.StreamRecordSerializer.deserialize(StreamRecordSerializer.java:110)
    at org.apache.flink.streaming.runtime.streamrecord.StreamRecordSerializer.deserialize(StreamRecordSerializer.java:41)
    at org.apache.flink.runtime.plugable.NonReusingDeserializationDelegate.read(NonReusingDeserializationDelegate.java:55)
    at org.apache.flink.runtime.io.network.api.serialization.SpillingAdaptiveSpanningRecordDeserializer.getNextRecord(SpillingAdaptiveSpanningRecordDeserializer.java:125)
    at org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput(StreamInputProcessor.java:127)
    at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run(OneInputStreamTask.java:56)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:172)
    at org.apache.flink.runtime.taskmanager.Task.run(Task.java:581)
    at java.lang.Thread.run(Thread.java:745)

我如何UnmodifiableCollection使用 Flink？

4

2 回答 2

18

问题是CollectionSerializerKryo 的默认值不能再次反序列化集合，因为它不可修改（.add()调用失败）。

为了解决这个问题，我们可以使用UnmodifiableCollectionsSerializer来自kryo-serializers项目的。Flink 传递依赖于项目，因此无需将其添加为依赖项。

接下来，我们必须向 Flink 的 Kryo 实例注册序列化程序。

StreamExecutionEnvironment see = StreamExecutionEnvironment.getExecutionEnvironment();
Class<?> unmodColl = Class.forName("java.util.Collections$UnmodifiableCollection");
see.getConfig().addDefaultKryoSerializer(unmodColl, UnmodifiableCollectionsSerializer.class);

通常，我们不必调用Class.forName()注册序列化程序，但在这种情况下，java.util.Collections$UnmodifiableCollection包是可见的，因此我们不能直接访问该类。

于 2015-09-08T08:38:17.160 回答
0

你可以试试这个，因为 kryo-serializers repo 在过去几年没有改变。

val env = StreamExecutionEnvironment.getExecutionEnvironment
val avroKryoSerializerUtil = new AvroKryoSerializerUtils
avroKryoSerializerUtil.addAvroSerializersIfRequired(env.getConfig,classOf[GenericData.Record])
于 2021-08-18T12:39:11.210 回答