Combined with case: the usage of the lowest API (processfunction) in Flink framework

summary

I mentioned in the previous article ,Flink The framework provides three layers API Complete stream processing task . So far, I have learned DataStream API ,ProcessFunction API  yes Flink The bottom of the middle API, You can access timestamps 、watermark  And register timed events . You can also output specific events .、

Process Function  Used to build event driven applications and implement custom business logic , If the window function and the conversion operator cannot meet the requirements of the business , Please come out ProcessFunction  To complete the development task .Flink SQL  Is the use of  Process Function  Realized .

Flink  Provides  8  individual  Process Function as follows ：

ProcessFunction

、

KeyedProcessFunction

、

CoProcessFunction

、

ProcessJoinFunction

、

BroadcastProcessFunction

、

KeyedBroadcastProcessFunction

、

ProcessWindowFunction

、

ProcessAllWindowFunction

. So let's start with KeyedProcessFunction For example .

KeyedProcessFunction<K, I, O>

It is mainly used to operate KeyedStream, It processes every element of the stream , Output is  0  individual 、1  One or more elements . be-all  Process Function  Inherit from RichFunction  Interface , So there are  open()、close() and  getRuntimeContext() Other methods . In addition, two methods are provided ：

• This method is called by every element in the data flow , The result of the call will be placed in  Collector  The output type in the data .Context  The timestamp that can access the element , Elemental key, as well as  TimerService  Time service .Context  You can also output the results to other streams (side outputs).

processElement(I value, Context ctx, Collector<O> out)


• Called when a previously registered timer triggers . Parameters  timestamp  The trigger time stamp set for the timer .Collector Is the set of output results .OnTimerContext  and processElement  Of  Context  Same parameter , Provides some information about the context . For example, timer triggered time information ( Event time or processing time ).

onTimer(long timestamp, OnTimerContext ctx, Collector<O> out) 


Timer

Context  in  TimerService Summary of the target party ：

• Returns the current processing time

long currentProcessingTime() 


• Returns the current  watermark  The timestamp

long currentWatermark() 


• Register current  key  Of processing time  Timer for , When  processing time  When the time comes , Trigger  timer.

void registerProcessingTimeTimer(long timestamp)


• Register current  key  Of  event time  Timer . When the water level is greater than or equal to the timer registration time , Trigger timer to execute callback function .

void registerEventTimeTimer(long timestamp) 


• Before deleting the register processing time timer . If there is no timer for this timestamp , Do not perform

void deleteProcessingTimeTimer(long timestamp)


• Delete previously registered event time timers , If there is no timer for this timestamp , Do not perform .

void deleteEventTimeTimer(long timestamp)


Test code ：

public class ProcessTest1_KeyedProcessFunction {
 public static void main(String[] args) throws Exception{
 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 env.setParallelism(1);

 DataStreamSource<String> inputStream = env.socketTextStream(&quot;localhost&quot;, 7777);
 DataStream<SensorReading> dataStream = inputStream.map(line -> {
 String[] fields = line.split(&quot;,&quot;);
 return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
 });

 // test keyedprocessFunction  Group first , Custom processing
 dataStream.keyBy(&quot;id&quot;)
 .process(new MyProcess())
 .print();


 env.execute();
 }

 //  Implement custom processing functions
 public static class MyProcess extends KeyedProcessFunction<Tuple,SensorReading,Integer> {

 ValueState<Long> tsTimeState;

 @Override
 public void open(Configuration parameters) throws Exception {
 tsTimeState = getRuntimeContext().getState(new ValueStateDescriptor<Long>(
 &quot;tsTimeState&quot;,Long.class
 ));
 }

 @Override
 public void processElement(SensorReading value, Context ctx, Collector<Integer> out) throws Exception {
 out.collect(value.getId().length());

 // Context operation
 ctx.timestamp();
 ctx.getCurrentKey();
 //  Sidestream
 //ctx.output();
 //  Get the current system processing time
 ctx.timerService().currentProcessingTime();
 //  Get the current event time
 ctx.timerService().currentWatermark();
 //  Register the system processing time timer
 ctx.timerService().registerProcessingTimeTimer( ctx.timerService().currentProcessingTime() + 1000L);
 tsTimeState.update( ctx.timerService().currentProcessingTime() + 1000L);

 //  Register the event time timer
 //ctx.timerService().registerEventTimeTimer((value.getTimestamp() + 10) * 1000L);
 //  Delete time
 //ctx.timerService().deleteProcessingTimeTimer(tsTimeState.value());
 }

 @Override
 public void onTimer(long timestamp, OnTimerContext ctx, Collector<Integer> out) throws Exception {
 System.out.println(timestamp+&quot; Timer triggered &quot;);

 ctx.getCurrentKey();
 //ctx.output();
 ctx.timeDomain();
 }

 @Override
 public void close() throws Exception {
 tsTimeState.clear();
 }
 }
}


A case ：

Monitor the temperature value of the temperature sensor , If the temperature value is  10  Rise continuously within seconds , Call the police .

public class ProcessTest2_ApplicationCase {
 public static void main(String[] args) throws Exception {
 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 env.setParallelism(1);

 DataStreamSource<String> inputStream = env.socketTextStream(&quot;localhost&quot;, 7777);
 DataStream<SensorReading> dataStream = inputStream.map(line -> {
 String[] fields = line.split(&quot;,&quot;);
 return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
 });


 dataStream.keyBy(&quot;id&quot;)
 .process(new TempConsIncreWarring(10))
 .print();

 env.execute();
 }

 //  Custom function   Test for a period of time （ Time domain ） The internal temperature rises continuously , Output alarm
 private static class TempConsIncreWarring extends KeyedProcessFunction<Tuple, SensorReading, String> {
 //  Define time fields
 private Integer interval;

 public TempConsIncreWarring(Integer interval) {
 this.interval = interval;
 }

 //  Define the State , Save the last temperature value , Timer timestamp
 private ValueState<Double> lastTempState;
 private ValueState<Long> tsTimeState;

 @Override
 public void open(Configuration parameters) throws Exception {
 lastTempState = getRuntimeContext().getState(new ValueStateDescriptor<Double>(
 &quot;lastTempState&quot;, Double.class, Double.MIN_VALUE));
 tsTimeState = getRuntimeContext().getState(new ValueStateDescriptor<Long>(
 &quot;tsTimeState&quot;, Long.class));
 }

 @Override
 public void processElement(SensorReading value, Context ctx, Collector<String> out) throws Exception {
 //  Removal status
 Double lastTemp = lastTempState.value();
 Long tsTime = tsTimeState.value();

 //  Register if the temperature rises 10 Seconds later and when there is no timer , wait for
 if (value.getTemperature() > lastTemp && tsTime == null) {
 //  Calculate the timer timestamp
 Long ts = ctx.timerService().currentProcessingTime() + interval * 1000L;
 //  Register timer
 ctx.timerService().registerProcessingTimeTimer(ts);
 //  Update time status
 tsTimeState.update(ts);
 }
 //  If the temperature drops   The timer needs to be deleted
 if (value.getTemperature() < lastTemp && tsTime != null) {
 ctx.timerService().deleteProcessingTimeTimer(tsTime);
 //  Clear the time status timer
 tsTimeState.clear();
 }

 //  Update temperature status
 lastTempState.update(value.getTemperature());
 }

 @Override
 public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
 //  Timer triggered   Output office reporting information
 out.collect(&quot; sensor &quot;+ ctx.getCurrentKey().getField(0) + &quot; The temperature value is continuous &quot;+ interval +&quot; Seconds have been rising &quot;);
 tsTimeState.clear();
 }

 @Override
 public void close() throws Exception {
 lastTempState.clear();
 }
 }
}


Running results ：

Side flow output

The side stream output function can generate multiple streams , And the data types of these streams can be different . A sidestream can be defined as  OutputTag[X] object ,X  Is the data type of the output stream .

A case ：

Monitor the sensor temperature value , Set the temperature below  30  Degree data output to  SideOutput

public class ProcessTest3_SideOutputCase {
 public static void main(String[] args) throws Exception {
 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 env.setParallelism(1);

 DataStreamSource<String> inputStream = env.socketTextStream(&quot;localhost&quot;, 7777);

 DataStream<SensorReading> dataStream = inputStream.map(line -> {
 String[] fields = line.split(&quot;,&quot;);
 return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
 });


 //  Definition outputTag  Indicates low temperature flow
 OutputTag lowTemp = new OutputTag<SensorReading>(&quot;lowTemp&quot;) {
 };

 //  Customize the measurement output flow to realize shunting operation
 SingleOutputStreamOperator<SensorReading> highTempStream = dataStream.process(new ProcessFunction<SensorReading, SensorReading>() {

 @Override
 public void processElement(SensorReading value, Context ctx, Collector<SensorReading> out) throws Exception {
 //  Judge that the temperature is greater than 30  High temperature flow   Output to the mainstream   The low temperature flow is output in the side flow
 if (value.getTemperature() > 30) {
 out.collect(value);
 } else {
 ctx.output(lowTemp, value);
 }
 }
 });

 highTempStream.print(&quot;high-Temp&quot;);
 highTempStream.getSideOutput(lowTemp).print(&quot;low&quot;);


 env.execute();
 }
}


Running results ：