38、Flink 的 WindowAssigner 之 GlobalWindows 示例
1、注意使用 GlobalWindows 需要自定义 Trigger，否则窗口中的数据不会被计算。
2、代码示例
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;
public class _05_WindowAssignerGlobal {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<String> input = env.socketTextStream("localhost", 8888);
// 此窗口模式仅在指定了自定义的 trigger 时有用，否则计算不会发生，因为全局窗口没有天然的终点去触发其中积累的数据
input
.keyBy(e -> e)
// 多并行 Task
.window(GlobalWindows.create())
.trigger(new Trigger<String, GlobalWindow>() {
@Override
public TriggerResult onElement(String s, long l, GlobalWindow globalWindow, TriggerContext triggerContext) throws Exception {
return null;
}
@Override
public TriggerResult onProcessingTime(long l, GlobalWindow globalWindow, TriggerContext triggerContext) throws Exception {
return null;
}
@Override
public TriggerResult onEventTime(long l, GlobalWindow globalWindow, TriggerContext triggerContext) throws Exception {
return null;
}
@Override
public void clear(GlobalWindow globalWindow, TriggerContext triggerContext) throws Exception {
}
})
.apply(new WindowFunction<String, String, String, GlobalWindow>() {
@Override
public void apply(String s, GlobalWindow globalWindow, Iterable<String> iterable, Collector<String> collector) throws Exception {
for (String res : iterable) {
collector.collect(res);
}
}
})
.print();
env.execute();
}
}
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
页:
[1]