Flink模拟项目: 计算最热门Top N商品
作者:互联网
为了统计每个窗口下最热门的商品,我们需要再次按窗口进行分组,这里根据ItemViewCount中的windowEnd进行keyBy()操作。然后使用ProcessFunction实现一个自定义的TopN函数TopNHotItems来计算点击量排名前3名的商品,并将排名结果格式化成字符串,便于后续输出。
.keyBy("windowEnd") .process(new TopNHotItems(3)); // 求点击量前3名的商品
ProcessFunction是Flink提供的一个low-level API,用于实现更高级的功能。它主要提供了定时器timer的功能(支持EventTime或ProcessingTime)。本案例中我们将利用timer来判断何时收齐了某个window下所有商品的点击量数据。由于Watermark的进度是全局的,在processElement方法中,每当收到一条数据ItemViewCount,我们就注册一个windowEnd+1的定时器(Flink框架会自动忽略同一时间的重复注册)。windowEnd+1的定时器被触发时,意味着收到了windowEnd+1的Watermark,即收齐了该windowEnd下的所有商品窗口统计值。我们在onTimer()中处理将收集的所有商品及点击量进行排序,选出TopN,并将排名信息格式化成字符串后进行输出。
这里我们还使用了ListState<ItemViewCount>来存储收到的每条ItemViewCount消息,保证在发生故障时,状态数据的不丢失和一致性。ListState是Flink提供的类似Java List接口的State API,它集成了框架的checkpoint机制,自动做到了exactly-once的语义保证。
// 求某个窗口中前 N 名的热门点击商品,key 为窗口时间戳,输出为 TopN 的结果字符串 class TopNHotItems(topSize: Int) extends KeyedProcessFunction[Tuple, ItemViewCount, String] { private var itemState : ListState[ItemViewCount] = _ override def open(parameters: Configuration): Unit = { super.open(parameters) // 命名状态变量的名字和状态变量的类型 val itemsStateDesc = new ListStateDescriptor[ItemViewCount]("itemState-state", classOf[ItemViewCount]) // 定义状态变量 itemState = getRuntimeContext.getListState(itemsStateDesc) } override def processElement(input: ItemViewCount, context: KeyedProcessFunction[Tuple, ItemViewCount, String]#Context, collector: Collector[String]): Unit = { // 每条数据都保存到状态中 itemState.add(input) // 注册 windowEnd+1 的 EventTime Timer, 当触发时,说明收齐了属于windowEnd窗口的所有商品数据 // 也就是当程序看到windowend + 1的水位线watermark时,触发onTimer回调函数 context.timerService.registerEventTimeTimer(input.windowEnd + 1) } override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Tuple, ItemViewCount, String]#OnTimerContext, out: Collector[String]): Unit = { // 获取收到的所有商品点击量 val allItems: ListBuffer[ItemViewCount] = ListBuffer() import scala.collection.JavaConversions._ for (item <- itemState.get) { allItems += item } // 提前清除状态中的数据,释放空间 itemState.clear() // 按照点击量从大到小排序 val sortedItems = allItems.sortBy(_.count)(Ordering.Long.reverse).take(topSize) // 将排名信息格式化成 String, 便于打印 val result: StringBuilder = new StringBuilder result.append("====================================\n") result.append("时间: ").append(new Timestamp(timestamp - 1)).append("\n") for(i <- sortedItems.indices){ val currentItem: ItemViewCount = sortedItems(i) // e.g. No1: 商品ID=12224 浏览量=2413 result.append("No").append(i+1).append(":") .append(" 商品ID=").append(currentItem.itemId) .append(" 浏览量=").append(currentItem.count).append("\n") } result.append("====================================\n\n") // 控制输出频率,模拟实时滚动结果 Thread.sleep(1000) out.collect(result.toString) } }
最后我们可以在main函数中将结果打印输出到控制台,方便实时观测:
.print();
至此整个程序代码全部完成,我们直接运行main函数,就可以在控制台看到不断输出的各个时间点统计出的热门商品。
2.2.5 完整代码
最终完整代码如下:
case class UserBehavior(userId: Long, itemId: Long, categoryId: Int, behavior: String, timestamp: Long) case class ItemViewCount(itemId: Long, windowEnd: Long, count: Long) object HotItems { def main(args: Array[String]): Unit = { val env = StreamExecutionEnvironment.getExecutionEnvironment env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) env.setParallelism(1) val stream = env .readTextFile("YOUR_PATH\\resources\\UserBehavior.csv") .map(line => { val linearray = line.split(",") UserBehavior(linearray(0).toLong, linearray(1).toLong, linearray(2).toInt, linearray(3), linearray(4).toLong) }) .assignAscendingTimestamps(_.timestamp * 1000) .filter(_.behavior=="pv") .keyBy("itemId") .timeWindow(Time.minutes(60), Time.minutes(5)) .aggregate(new CountAgg(), new WindowResultFunction()) .keyBy(1) .process(new TopNHotItems(3)) .print() env.execute("Hot Items Job") } // COUNT 统计的聚合函数实现,每出现一条记录加一 class CountAgg extends AggregateFunction[UserBehavior, Long, Long] { override def createAccumulator(): Long = 0L override def add(userBehavior: UserBehavior, acc: Long): Long = acc + 1 override def getResult(acc: Long): Long = acc override def merge(acc1: Long, acc2: Long): Long = acc1 + acc2 } // 用于输出窗口的结果 class WindowResultFunction extends WindowFunction[Long, ItemViewCount, Tuple, TimeWindow] { override def apply(key: Tuple, window: TimeWindow, aggregateResult: Iterable[Long], collector: Collector[ItemViewCount]) : Unit = { val itemId: Long = key.asInstanceOf[Tuple1[Long]].f0 val count = aggregateResult.iterator.next collector.collect(ItemViewCount(itemId, window.getEnd, count)) } } // 求某个窗口中前 N 名的热门点击商品,key 为窗口时间戳,输出为 TopN 的结果字符串 class TopNHotItems(topSize: Int) extends KeyedProcessFunction[Tuple, ItemViewCount, String] { private var itemState : ListState[ItemViewCount] = _ override def open(parameters: Configuration): Unit = { super.open(parameters) // 命名状态变量的名字和状态变量的类型 val itemsStateDesc = new ListStateDescriptor[ItemViewCount]("itemState-state", classOf[ItemViewCount]) // 从运行时上下文中获取状态并赋值 itemState = getRuntimeContext.getListState(itemsStateDesc) } override def processElement(input: ItemViewCount, context: KeyedProcessFunction[Tuple, ItemViewCount, String]#Context, collector: Collector[String]): Unit = { // 每条数据都保存到状态中 itemState.add(input) // 注册 windowEnd+1 的 EventTime Timer, 当触发时,说明收齐了属于windowEnd窗口的所有商品数据 // 也就是当程序看到windowend + 1的水位线watermark时,触发onTimer回调函数 context.timerService.registerEventTimeTimer(input.windowEnd + 1) } override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Tuple, ItemViewCount, String]#OnTimerContext, out: Collector[String]): Unit = { // 获取收到的所有商品点击量 val allItems: ListBuffer[ItemViewCount] = ListBuffer() import scala.collection.JavaConversions._ for (item <- itemState.get) { allItems += item } // 提前清除状态中的数据,释放空间 itemState.clear() // 按照点击量从大到小排序 val sortedItems = allItems.sortBy(_.count)(Ordering.Long.reverse).take(topSize) // 将排名信息格式化成 String, 便于打印 val result: StringBuilder = new StringBuilder result.append("====================================\n") result.append("时间: ").append(new Timestamp(timestamp - 1)).append("\n") for(i <- sortedItems.indices){ val currentItem: ItemViewCount = sortedItems(i) // e.g. No1: 商品ID=12224 浏览量=2413 result.append("No").append(i+1).append(":") .append(" 商品ID=").append(currentItem.itemId) .append(" 浏览量=").append(currentItem.count).append("\n") } result.append("====================================\n\n") // 控制输出频率,模拟实时滚动结果 Thread.sleep(1000) out.collect(result.toString) } } }
2.2.6 更换Kafka 作为数据源
实际生产环境中,我们的数据流往往是从Kafka获取到的。如果要让代码更贴近生产实际,我们只需将source更换为Kafka即可:
val properties = new Properties() properties.setProperty("bootstrap.servers", "localhost:9092") properties.setProperty("group.id", "consumer-group") properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer") properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer") properties.setProperty("auto.offset.reset", "latest") val env = StreamExecutionEnvironment.getExecutionEnvironment env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) env.setParallelism(1) val stream = env .addSource(new FlinkKafkaConsumer[String]("hotitems", new SimpleStringSchema(), properties))
当然,根据实际的需要,我们还可以将Sink指定为Kafka、ES、Redis或其它存储,这里就不一一展开实现了。
标签:windowEnd,String,ItemViewCount,Top,Flink,热门,Long,val,def 来源: https://www.cnblogs.com/tesla-turing/p/13276471.html