与apache flink一起使用collections$unmodifiablecollection

ql3eal8s 于 2021-06-24 发布在 Flink

关注(0)|答案(1)|浏览(668)

使用带有以下代码的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在一起？

apache-flink Kryo

来源：https://stackoverflow.com/questions/32453030/using-an-collectionsunmodifiablecollection-with-apache-flink

1条答案

按热度按时间

pod7payv1#

问题是默认值 CollectionSerializer 的kryo无法再次反序列化集合，因为它不可修改（ .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 包是可见的，所以我们不能直接访问类。

赞(0）回复(0）举报 2021-06-24

我来回答

与apache flink一起使用collections$unmodifiablecollection

1条答案

相关问题

热门标签

最新问答