我们之前学习的转换算子是无法访问事件的时间戳信息和水位线信息的。而这在一些应用场景下，极为重要。例如：MapFunction 这样的 map 转换算子就无法访问时间戳或当前事件的事件时间；
基于此，DataStreamAPI提供了一系列的底层API转换算子，可以访问时间戳、Watermark以及注册定时事件。还可以输出特定的一些事件，例如：超时事件等。
ProcessFunction用来构建事件驱动的应用以及实现自定义的业务逻辑；Flink提供了8个ProcessFunction：

  • ProcessFunction ：常用
  • KeyedProcessFunction ：常用
  • CoProcessFunction
  • ProcessJoinFunction
  • BroadcastProcessFunction
  • KeyedBroadcastProcessFunction
  • ProcessWindowFunction
  • ProcessAllWindowFunction

一、KeyedProcessFunction

用来操作 KeyedStream。KeyedProcessFunction 会处理流的每一个元素，输出为0个、1个或者多个元素。所有的ProcessFunction 都继承自 RichFunction接口，所以都有 open()、close() 和 getRuntimeContext() 等方法。而 KeyedProcessFunction 还提供了额外两个方法：

  • processElement(v: IN, ctx: Context, out: Collector[OUT])，流中的每一个元素都会调用这个方法，调用结果将会放在 Collector 数据类型中输出。Context 可以访问元素的时间戳、元素的key 以及TimeService 时间服务。Context 还可以将结果输出到别的流(side outputs)
  • onTimer(timestamp: Long, ctx: OnTimerContext, out: Collector[OUT]) 是一个回调函数。当之前注册的定时器触发时调用。参数 timestamp 为定时器所设定的触发的时间戳。 Collector 为输出的结果集合。OnTimerContext 和 processElement 的Context参数一样，提供了上下文的一些信息；

二、TimerService和定时器(Timer)

实例：

  1. package com.wells.flink.demo.process;
  3. import org.apache.flink.api.common.functions.MapFunction;
  4. import org.apache.flink.api.common.state.ValueState;
  5. import org.apache.flink.api.common.state.ValueStateDescriptor;
  6. import org.apache.flink.api.java.tuple.Tuple;
  7. import org.apache.flink.api.java.tuple.Tuple2;
  8. import org.apache.flink.configuration.Configuration;
  9. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  10. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  11. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  12. import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
  13. import org.apache.flink.util.Collector;
  15. /**
  16.  * Description ProcessAPI
  17.  * Created by wells on 2020-05-24 17:45:41
  18.  */
  20. public class ProcessAPITest {
  21.     public static void main(String[] args) throws Exception {
  22.         String host = "localhost";
  23.         int port = 9999;
  25.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  26.         env.setParallelism(1);
  28.         DataStreamSource<String> dataStreamSource = env.socketTextStream(host, port);
  29.         dataStreamSource.print("socketSource");
  30.         SingleOutputStreamOperator<String> processStreamSource = dataStreamSource.map(new MapFunction<String, Tuple2<String, Integer>>() {
  31.             @Override
  32.             public Tuple2<String, Integer> map(String line) throws Exception {
  33.                 String[] split = line.split(" ", -1);
  34.                 return new Tuple2<String, Integer>(split[0], Integer.parseInt(split[1]));
  35.             }
  36.         }).keyBy(0).process(new MyKeyedProcessFunction());
  37.         processStreamSource.print("processSource");
  39.         env.execute();
  40.     }
  41. }
  43. class MyKeyedProcessFunction extends KeyedProcessFunction<Tuple, Tuple2<String, Integer>, String> {
  44.     // 定义一个状态，用来保存上一行记录的数字
  45.     private transient ValueState<Integer> lastNumber;
  47.     // 记录增加的定时器的时间，方便清除
  48.     private transient ValueState<Long> currentTimerTs;
  50.     @Override
  51.     public void open(Configuration parameters) throws Exception {
  52.         lastNumber = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("lastNumber", Integer.class, 0));
  53.         currentTimerTs = getRuntimeContext().getState(new ValueStateDescriptor<Long>("currentTimerTs", Long.class, 0L));
  54.     }
  56.     // 处理流中的每一个元素
  57.     @Override
  58.     public void processElement(Tuple2<String, Integer> value, Context ctx, Collector<String> out) throws Exception {
  59.         // 得到上一次的状态值
  60.         Integer preNumber = lastNumber.value();
  62.         // 更新上一次的状态值
  63.         lastNumber.update(value.f1);
  65.         // 当前定时器的时间
  66.         Long currTimerTs = currentTimerTs.value();
  68.         // 如果数字减小或者第一次设置定时器，则触发定时器
  69.         if (value.f1 < preNumber && currTimerTs == 0) {
  70.             // 触发定时器时间: 在当前时间基础上加10s 即 当前时间过10s后数字下降，触发定时器
  71.             long timerTs = ctx.timerService().currentProcessingTime() + 10000L;
  72.             ctx.timerService().registerProcessingTimeTimer(timerTs);
  73.             currentTimerTs.update(timerTs);
  74.         } else if (value.f1 > preNumber || preNumber == 0) {
  75.             // 数字变大 或者 第一次进入 则删除定时器
  76.             ctx.timerService().deleteProcessingTimeTimer(currTimerTs);
  77.             currentTimerTs.clear();
  78.         }
  79.     }
  81.     // 定时器要做的事情
  82.     @Override
  83.     public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
  84.         out.collect("number more than last number");
  85.         currentTimerTs.clear();
  86.     }
  87. }

三、侧输出流

通过在processFunction处理将流分开；也可以使用 split 算子

  1. package com.wells.flink.demo.process;
  3. import org.apache.flink.api.common.functions.MapFunction;
  4. import org.apache.flink.api.common.typeinfo.TypeInformation;
  5. import org.apache.flink.configuration.Configuration;
  6. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  7. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  8. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  9. import org.apache.flink.streaming.api.functions.ProcessFunction;
  10. import org.apache.flink.util.Collector;
  11. import org.apache.flink.util.OutputTag;
  13. /**
  14.  * Description 侧输出流，通过
  15.  * Created by wells on 2020-05-24 19:06:57
  16.  */
  18. public class SideOutputTest {
  19.     public static void main(String[] args) throws Exception {
  20.         String host = "localhost";
  21.         int port = 9999;
  23.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  24.         env.setParallelism(1);
  26.         DataStreamSource<String> dataStreamSource = env.socketTextStream(host, port);
  27.         dataStreamSource.print("socketSource");
  28.         SingleOutputStreamOperator<String> processStreamSource = dataStreamSource.map(new MapFunction<String, Integer>() {
  29.             @Override
  30.             public Integer map(String line) throws Exception {
  31.                 return Integer.parseInt(line);
  32.             }
  33.         }).process(new MyProcessFunction());
  34.         processStreamSource.print("processSource");
  35.         processStreamSource.getSideOutput(new OutputTag<String>("numberDecrease", TypeInformation.of(String.class)))
  36.                 .print("numberDecrease");
  38.         env.execute();
  39.     }
  40. }
  42. class MyProcessFunction extends ProcessFunction<Integer, String> {
  43.     private transient OutputTag<String> outputTag;
  45.     @Override
  46.     public void open(Configuration parameters) throws Exception {
  47.         outputTag = new OutputTag<String>("numberDecrease", TypeInformation.of(String.class));
  48.     }
  50.     // 处理流中的每一个元素
  51.     @Override
  52.     public void processElement(Integer value, Context ctx, Collector<String> out) throws Exception {
  53.         // 如果数字减小或者第一次设置定时器，则触发定时器
  54.         if (value < 0) {
  55.             ctx.output(outputTag, String.valueOf(value));
  56.         } else {
  57.             out.collect(String.valueOf(value));
  58.         }
  59.     }
  60. }