Flink：状態を保存して別のストリームで使用する方法は？

Question

私は、ファイルから情報を読み取り、各行を保存し、この状態を使って別のストリームをフィルタリングする必要があるFlinkのユースケースを持っています。

connectオペレーターとRichCoFlatMapFunctionではこれがすべて機能していますが、複雑すぎると感じています。また、私は状態のすべてがファイルからロードされる前にflatMap2が実行を開始できることを心配：負荷状態のこのパターンを達成するために

fileStream 
    .connect(partRecordStream.keyBy((KeySelector<PartRecord, String>) partRecord -> partRecord.getPartId())) 
    .keyBy((KeySelector<String, String>) partId -> partId, (KeySelector<PartRecord, String>) partRecord -> partRecord.getPartId()) 
    .flatMap(new RichCoFlatMapFunction<String, PartRecord, PartRecord>() { 
     private transient ValueState<String> storedPartId; 
     @Override 
     public void flatMap1(String partId, Collector<PartRecord> out) throws Exception { 
      // store state 
      storedPartId.update(partId); 
     } 

     @Override 
     public void flatMap2(PartRecord record, Collector<PartRecord> out) throws Exception { 
      if (record.getPartId().equals(storedPartId.value())) { 
       out.collect(record); 
      } else { 
       // do nothing 
      } 
     } 

     @Override 
     public void open(Configuration parameters) throws Exception { 
      ValueStateDescriptor<String> descriptor = 
        new ValueStateDescriptor<>(
          "partId", // the state name 
          TypeInformation.of(new TypeHint<String>() {}), 
          null); 
      storedPartId = getRuntimeContext().getState(descriptor); 
     } 
    }); 

は（FLINK 1.1.3のように）より良い方法があり、その後のストリームでそれを使用しますか？

Answer 1

CoFlatMapFunctionについてのご意見は正しいですか？ flatMap1とflatMap2が呼び出される順序は制御できず、データの到着順序に依存します。したがって、すべてのデータがflatMap1によって読み取られる前にflatMap2が呼び出される可能性があります。

ストリームの処理を開始する前にすべてのデータを読み取る唯一の方法は、メソッドのデータをRichFlatMapFunctionで消費することです。つまり、手動でファイルを読み込んで解析する必要があります。

これは基本的にブロードキャスト結合戦略です。つまり、オペレータの各並列インスタンスがこれを行います。欠点は、ファイルのデータが複製されることです。メリットは、「メイン」ストリームをシャッフルする必要がないことです（keyBy()を使用する必要はありません）。