DataStreamプログラムでユーザー機能を設定する方法は？

Question

コンフィグレーションでストリーミング環境を作成し、RichMapFunctionのopen()メソッドでこのコンフィグレーションにアクセスしようとしました。

例：open()メソッドをデバッグするとき

Configuration conf = new Configuration(); 
    conf.setBoolean("a", true); 
    StreamExecutionEnvironment env = 
     StreamExecutionEnvironment.createLocalEnvironment(8, conf); 

    DataStreamSource<Integer> source = env.fromElements(5,5,5,5,5); 
    source.map(new RichMapFunction<Integer, Integer>() { 

     @Override 
     public void open(Configuration parameters) throws Exception { 
      boolean a = parameters.getBoolean("a", false); 
      super.open(parameters); 
     } 

     @Override 
     public Integer map(Integer value) throws Exception { 
      return value; 
     } 
    }).print(); 

    env.execute(); 

は、しかし、私は設定が空であることがわかります。

私は間違っていますか？ストリーミング環境のRichFunctionに設定を正しく渡すにはどうすればよいですか？

Answer 1

FlinkのDataStream APIとDataSet APIは、同じ例ではRichMapFucntionのようなユーザー関数インターフェイスを共有しています。

openメソッドのFlinkのRichFunctionのパラメータは、DataSet APIの最初のバージョンのレガシーであり、DataStream APIでは使用されていません。 Flinkは、map()コールで指定したオブジェクトをシリアル化し、それをパラレルワーカーに送ります。したがって、オブジェクト内のパラメータを通常のフィールドとして直接設定することができます。