我有一个简单的代码如下,当我在ide中运行它时,控制台上没有打印任何东西,有人能帮我看一下吗?谢谢
import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
object WindowTest {
def main(args: Array[String]): Unit = {
val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)
val ds = env.fromElements(
(1, "a"), (2, "b"), (3, "c"), (4, "e"), (5, "f"), (6, "g"), (7, "h"), (8, "g"), (1, "1a"), (2, "2b"), (3, "3c"), (4, "4e"), (5, "5f"), (6, "g"), (7, "h"), (8, "g")
)
val ds2 = ds.keyBy(0).timeWindow(Time.seconds(10))
.process(new ProcessWindowFunction[(Int, String), String, Tuple, TimeWindow] {
override def process(key: Tuple, context: Context, elements: Iterable[(Int, String)], out: Collector[String]): Unit = {
val k = key.getField[Int](0)
val w = context.window
val start = w.getStart
val end = w.getEnd
val hc = context.window.hashCode()
//NOT CALLED
println(s"k=$k,start=$start, end=$end,hc=$hc")
}
})
ds2.print()
env.execute()
Thread.sleep(30 * 1000)
}
}
1条答案
按热度按时间vaqhlq811#
processwindowfunction从未被调用,因为窗口从未被触发。它从未被触发,因为它在几毫秒内运行到完成(粗略地说),所以它不太可能在系统时钟时间正好处于10秒边界的时刻运行,而这正是触发此处理时间窗口所必须发生的。