Flink测流输出

public class SideOutputDemo { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStreamSource lines = env.socketTextStream("linux01", 7777); //需指定泛型的类型 或者new OutputTag的子类(即使用匿名内部类) //OutputTag oddTag = new OutputTag("odd-data") {}; OutputTag oddTag = new OutputTag<>("odd-data", Types.INT); OutputTag evenTag = new OutputTag<>("even-data",Types.INT); OutputTag strTag = new OutputTag<>("str-data",Types.STRING); SingleOutputStreamOperator mainStream = lines.process(new ProcessFunction() { @Override public void processElement(String line, Context ctx, Collector out) throws Exception { try { int num = Integer.parseInt(line); if (num % 2 == 0) { //偶数 ctx.output(oddTag, num); } else { //奇数 ctx.output(evenTag, num); } } catch (NumberFormatException e) { //字符串 ctx.output(strTag, line); } } }); DataStream oddStream = mainStream.getSideOutput(oddTag); DataStream strStream = mainStream.getSideOutput(strTag); oddStream.print("偶数流"); strStream.print("字符流"); mainStream.print("主流"); env.execute(); }}

public class GetWindowLateDataDemo { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //开启checkpointing env.enableCheckpointing(10000); //设置重启策略 env.setRestartStrategy(RestartStrategies.fixedDelayRestart(5, 5000)); DataStreamSource lines = env.socketTextStream("linux01", 7777); //获取Watermark SingleOutputStreamOperator linesWithWatermark = lines.assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(0)).withTimestampAssigner(new SerializableTimestampAssigner() { @Override public long extractTimestamp(String element, long recordTimestamp) { return Long.parseLong(element.split(",")[0]); } })); SingleOutputStreamOperator> wordAndCount = linesWithWatermark.map(new MapFunction>() { @Override public Tuple2 map(String line) throws Exception { if (line.startsWith("error")){ throw new RuntimeException("错误数据"); } String[] fields = line.split(","); return Tuple2.of(fields[1], Integer.parseInt(fields[2])); } }); //按照单词keyBy KeyedStream, String> keyedStream = wordAndCount.keyBy(tp -> tp.f0); //划分窗口 WindowedStream, String, TimeWindow> windowedStream = keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(5))); //给迟到的数据打上标签 OutputTag> lateDataTag = new OutputTag>("late-data"){}; windowedStream.sideOutputLateData(lateDataTag); SingleOutputStreamOperator> result = windowedStream.sum(1); //获取迟到的数据 DataStream> lateDataStream = result.getSideOutput(lateDataTag); result.print(); lateDataStream.print("迟到的数据"); env.execute(); }}