关于 flink addSink(new MySink ) 问题咨询 #Flink

关于 flink addSink(new MySink ) 问题咨询

val resDs = ds .map(new DwRichMapFunction) .keyBy(_.tableName) .timeWindowAll(Time.seconds(5)) .process(new DwProcessAllWindowFunction) .addSink(new MySink)

class MySink extends RichSinkFunction[String] {

override def open(parameters: Configuration): Unit = {   println("===========sink open ===========") }
override def invoke(elements: List[String], context: SinkFunction.Context[_]): Unit = {}
override def close(): Unit = {}

}

问题一： 为什么在启动的时候 会进入 MySink 的open 4次？ 问题二： 启动后的每一个window 会不会再次进入 MySink的 open 方法， 目测现在日志打印好像没有； 只进入 invoke 方法 #Flink