上文我们介绍了五种 KeyedState 的特性以及如何初始化,本文基于五种 KeyedState,通过一个案例向大家同时介绍五种 KeyedState 使用方法。
- ValueState
- ListState
- ReducingState
- AggregateState
- MapState
- // 用户浏览行为
- case class Event(user: String, url: String, timeStamp: Long)
数据类型为 case class Event,其中只包含3个字段,分别代表:
user - 用户 ID
url - 用户浏览 URL
timeStamp - 用户浏览时间
数据源通过 Random 随机生成用户 ID 与 URL,并结合 System.currentTimeMillis 作为 EventTime 发出,每隔1s发送一个 Event 类模拟用户浏览情况:
- import org.apache.flink.streaming.api.functions.source.SourceFunction
-
- import scala.util.Random
-
- class ClickHouse extends SourceFunction[Event] {
-
- var running: Boolean = true
-
- val ran