Flink DataStream 有没有类似 mapPartition 的 api？答案

【问题标题】：Does Flink DataStream have api like mapPartition?Flink DataStream 有没有类似 mapPartition 的 api？
【发布时间】：2020-03-03 18:13:23
【问题描述】：

我想像这样在stream.map() 中使用不可序列化的对象

stream.map { i =>
  val obj = new SomeUnserializableClass()
  obj.doSomething(i)
}

效率很低，因为我创建了很多SomeUnserializableClass 实例。实际上，它只能在每个工人中创建一次。

在 Spark 中，我可以使用 mapPartition 来执行此操作。但是在flink stream api中，我不知道。

【问题讨论】：

标签： apache-flink flink-streaming

【解决方案1】：

如果你正在处理一个不可序列化的类，我建议你创建一个RichFunction。在您的情况下是 RichMapFunction。

Flink 中的 Rich operator 有一个 open 方法，该方法在 taskmanager 中只作为初始化器执行一次。

所以诀窍是让你的字段瞬态并在你的 open 方法中实例化它。

查看以下示例：

public class NonSerializableFieldMapFunction extends RichMapFunction {

    transient SomeUnserializableClass someUnserializableClass;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        this.someUnserializableClass = new SomeUnserializableClass();
    }

    @Override
    public Object map(Object o) throws Exception {
        return someUnserializableClass.doSomething(o);
    }
}

那么您的代码将如下所示：

stream.map(new NonSerializableFieldMapFunction())

P.D：我使用的是 java 语法，请适应 scala。

【讨论】：