代码如下:
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows
import org.apache.flink.streaming.api.windowing.triggers.CountTrigger;
val person = senv.fromElements(("1","小张"),("2","小刘"),("3","小力"),("4","小心"))
val money = senv.fromElements(("1",100),("2",200),("3",300))
person.join(money).where(_._1).equalTo(_._1).window(GlobalWindows.create()).trigger(CountTrigger.of(2)).apply((x,y)=>{println( x+"==="+y)}).print()
senv.execute()
实验结果:
(1,小张)===(1,100)
1> ()
(2,小刘)===(2,200)
1> ()
(3,小力)===(3,300)
1> ()
没有输出结果的原因:
trigger(CountTrigger.of(2))没写
来自张云云．