假设我有一个带有原始状态的自定义rich函数。当flink作业结束时，如何将状态（从操作符的每个并行示例）返回到主/驱动程序代码？

abstract class MyRichMap extends RichMapFunction[SomeType, Unit] {

  protected var someVar: Engine = _ 

  override def open(parameters: Configuration): Unit = {
    // assume someVar inititation here
    ....
  }

  override def map(value: SomeType): Unit = {

    engine.process(value)

  }

val env = StreamExecutionEnvironment.getExecutionEnvironment
    ...
    someSource.map (new MyRichMap())

    env.execute()
    // How to get engine or some field of it here? (e.g., engine.someCounter)

最好的方法是什么？