Flink DataStream是否有类似mapPartition的api？

Question

我想像这样在stream.map（）中使用一个不可序列化的对象

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

它非常低效，因为我创建了许多一些UnseralizableClass实例。实际上，它只能在每个工作器中创建一次。

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

Answer 1

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

Flink中的富操作符有一个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。