非アクティブキーのクリーンアップFlinkストリームの状態はどのようになりますか？

Question

私の目標は、IDがイベントから抽出された最後のN idsを保持するFlinkストリーミングプログラムを持つことです。シンクはCassandraストアであるため、いつでもIDのリストを取得できます。カッサンドラはあらゆるイベントの直後に更新されることが重要です。

これはmapWithState（以下のコードを参照）で簡単に実装できます。しかし、このコードには重要な問題があります。状態にはuseridが入力されています。一部のユーザーはしばらくの間アクティブになっている可能性があります。私が心配しているのは、状態ストレージが永遠に成長するということです。

アクティブでないキーのクリーンアップ状態はどのようになりますか？

case class MyEvent(userId: Int, id: String) 

env 
    .addSource(new FlinkKafkaConsumer010[MyEvent]("vips", new MyJsonDeserializationSchema(), kafkaConsumerProperties)) 
    .keyBy(_.userId) 
    .mapWithState[(Int, Seq[String]), Seq[String]] { (in: MyEvent, currentIds: Option[Seq[String]]) => 
    val keepNIds = currentIds match { 
     case None => Seq(in.id) 
     case Some(cids) => (cids :+ in.id).takeRight(100) 
    } 
    ((in.userId, keepNIds), Some(keepNIds)) 
    } 
    .addSink { in: (Int, Seq[String]) => 
    CassandraSink.appDatabase.idsTable.store(...) 
    } 

Answer 1

成長状態は重要かつ正確な観察です。あなたのキースペースが動いているなら、これは間違いなく起こります。

Flink 1.2.0がこの問題を解決するProcessFunctionを追加しました。 ProcessFunctionはFlatMapFunctionに似ていますが、タイマーサービスにアクセスできます。コールバック関数が期限切れになると、onTimer()コールバック関数を呼び出すタイマーを登録できます。コールバックを使用して状態をクリーンアップすることができます。