一个清理闪烁流状态如何为非活动键？

Question

我的目标是有一个Flink流程序，保留最后的N个id，其中id是从事件中提取的。接收器是一个Cassandra存储区，因此可以随时获取ID列表。重要的是，卡桑德拉在每一次事件发生时都要立即得到最新消息。

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()回调函数。回调可用于清理状态。