我尝试在flink中实现窗口触发器,如果平均值高于阈值,它将触发。
流式数据将学生的姓名和标记用分隔符隔开 ,
. 如果学生的平均分数超过 90
不考虑尝试次数。
示例数据:
Fred,88
Fred,91
Wilma,93
.
.
当前flink代码:
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.triggers.Trigger.TriggerContext
import org.apache.flink.streaming.api.windowing.triggers.{CountTrigger, PurgingTrigger, Trigger, TriggerResult}
import org.apache.flink.streaming.api.windowing.windows.{GlobalWindow, Window}
case class Marks(name : String, mark : Double, count : Int)
class MarksTrigger[W <: Window] extends Trigger[Marks,W] {
override def onElement(element: Marks, timestamp: Long, window: W, ctx: TriggerContext): TriggerResult = {
if(element.mark > 90) TriggerResult.FIRE // fire if avg mark is > 90
else TriggerResult.CONTINUE
}
override def onProcessingTime(time: Long, window: W, ctx: TriggerContext): TriggerResult = {
TriggerResult.CONTINUE
}
override def onEventTime(time: Long, window: W, ctx: TriggerContext): TriggerResult = {
TriggerResult.CONTINUE
}
override def clear(window: W, ctx: TriggerContext) = ???
}
object Main {
def main(args: Array[String]) {
val env = StreamExecutionEnvironment.getExecutionEnvironment
val data = env.socketTextStream("localhost", 9999)
val fdata = data.map { values =>
val columns = values.split(",")
Marks(columns(0), columns(1).toDouble, 1)
}
val keyed = fdata.keyBy(_.name).
window(GlobalWindows.create()).
trigger(new MarksTrigger[GlobalWindow]()). // TODO
keyed.print()
env.execute()
}
}
正在计算平均值:在批处理模式下尝试了以下操作
case class Marks(name : String, mark : Double, count : Int)
val data = benv.fromElements(("Fred", 88.0), ("Fred", 95.0), ("Fred", 91.0), ("Wilma", 93.0), ("Wilma", 95.0), ("Wilma", 98.0))
data.map(x => (x._1, x._2, 1)).groupBy(0).reduce { (x, y) =>
(x._1, x._2 + y._2, x._3 + y._3)
}.map(x => Marks(x._1, x._2/x._3, x._3)).collect
我怎么把这些绑在一起?应该 .window()
以及 .trigger()
在计算平均数之前被调用或者平均数计算应该在里面做 onElement()
?
1条答案
按热度按时间eqfvzcg81#
我想出了解决办法