Flinkで非ブロッキングストリーミング

Question

こんにちは、次のように着信データを処理する必要があるFlinkジョブを実行しようとしています。 keyBy()の直後のプロセス演算子では、データのプロパティによっては時間がかかる場合があります。入ってくるデータが異なるID（ストリームに使用されています）を持っていても、処理機能の長い処理コードは他の入力データをブロックします。私はストリーム全体を意味します。 RuleProcessFunction.javaで

SingleOutputStreamOperator<Envelope> processingStream = deviceStream 
    .map(e -> (Envelope) e) 
    .keyBy((KeySelector<Envelope, String>) value -> value.eventId) // key by scenarios 
    .process(new RuleProcessFunction()); 

：

... 
@Override 
public void processElement(Envelope value, Context ctx, Collector<Envelope> out) throws Exception { 
    //handleEvent(value, ctx, out); 
    if (value.getEventId().equals("I")) { 
     System.out.println("hello i"); 
     for (long i = 0; i < 10000000000L; i++) { 

     } 
    } 
    out.collect(value); 
} 

私は長時間実行コードブロックは、ストリーム全体をブロックするべきではありません期待しています。私はIOの状況をブロックするためのAsyncFunctionがあることを知っていますが、私はそれがこのための正しい解決であるとは知らない。

Answer 1

Cassandraのような外部データベースからデータを取得しているわけではないので、私はAsyncFunctionを使う必要はないと思います。

あなたは、単一の並列処理でフリンクジョブを実行している可能性があります。並列処理を強化して、1つのコアがすべての処理とデータの受信を担当しないようにしてください。確かに、あなたがこれを行うなら、背圧が依然としてあるかもしれません。ソースからのデータの取り込みを担当するコアが、processFunctionを実行しているコアよりも速くデータを読み込んでいる場合、Flinkの背圧処理は取り込み速度を遅くします。