0
点赞
收藏
分享

微信扫一扫

结合案例总结Flink框架中的最底层API(ProcessFunction)用法

大家好,我是百思不得小赵。

创作时间:2022 年 5 月 30 日

概述

在之前总结的文章中有提到过,Flink框架提供了三层API完成流处理任务。至此已经学习了DataStream API ,ProcessFunction API 是Flink中最底层的API,可以访问时间戳、watermark 以及注册定时事件。还可以输出特定的一些事件。、
结合案例总结Flink框架中的最底层API(ProcessFunction)用法_大数据

Process Function 用来构建事件驱动的应用以及实现自定义的业务逻辑,若窗口函数以及转换算子都无法满足业务的要求时,需要请出ProcessFunction 去完成开发任务。Flink SQL 就是使用 Process Function 实现的。

Flink 提供了 8 个 Process Function如下:​​ProcessFunction​​​、​​KeyedProcessFunction​​​、​​CoProcessFunction​​​、​​ProcessJoinFunction​​​、​​BroadcastProcessFunction​​​、​​KeyedBroadcastProcessFunction​​​、​​ProcessWindowFunction​​​、​​ProcessAllWindowFunction​​。接下来我们以KeyedProcessFunction为例来进行学习。

KeyedProcessFunction<K, I, O>

它主要用来操作KeyedStream,会处理流的每一个元素,输出为 0 个、1 个或者多个元素。所有的 Process Function 都继承自RichFunction 接口,所以都有 open()、close()和 getRuntimeContext()等方法。除此之外还提供了两个方法:

  • 数据流中的每一个元素都会调用这个方法,调用结果将会放在 Collector 数据类型中输出。Context 可以访问元素的时间戳,元素的key,以及 TimerService 时间服务。Context 还可以将结果输出到别的流(side outputs)。

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

  • 当之前注册的定时器触发时调用。参数 timestamp 为定时器所设定的触发的时间戳。Collector为输出结果的集合。OnTimerContext 和processElement 的 Context 参数一样,提供了上下文的一些信息。例如定时器触发的时间信息(事件时间或者处理时间)。

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

定时器

Context 中 TimerService对象方汇总:

  • 返回当前处理时间

long currentProcessingTime()

  • 返回当前 watermark 的时间戳

long currentWatermark()

  • 注册当前 key 的processing time 的定时器,当 processing time 到达定时时间时,触发 timer。

void registerProcessingTimeTimer(long timestamp)

  • 注册当前 key 的 event time 定时器。当水位线大于等于定时器注册的时间时,触发定时器执行回调函数。

void registerEventTimeTimer(long timestamp)

  • 删除之前注册处理时间定时器。如果没有这个时间戳的定时器,则不执行

void deleteProcessingTimeTimer(long timestamp)

  • 删除之前注册的事件时间定时器,如果没有此时间戳的定时器,则不执行。

void deleteEventTimeTimer(long timestamp)

测试代码:

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

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

//测试keyedprocessFunction 先分组,自定义处理
dataStream.keyBy("id")
.process(new MyProcess())
.print();


env.execute();
}

// 实现自定义处理函数
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>(
"tsTimeState",Long.class
));
}

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

// Context操作
ctx.timestamp();
ctx.getCurrentKey();
// 侧流
//ctx.output();
// 获取当前系统处理时间
ctx.timerService().currentProcessingTime();
// 获取当前事件时间
ctx.timerService().currentWatermark();
// 注册系统处理时间定时器
ctx.timerService().registerProcessingTimeTimer( ctx.timerService().currentProcessingTime() + 1000L);
tsTimeState.update( ctx.timerService().currentProcessingTime() + 1000L);

// 注册事件时间定时器
//ctx.timerService().registerEventTimeTimer((value.getTimestamp() + 10) * 1000L);
// 删除时间
//ctx.timerService().deleteProcessingTimeTimer(tsTimeState.value());
}

@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<Integer> out) throws Exception {
System.out.println(timestamp+"定时器触发");

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

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

一个案例:

监控温度传感器的温度值,如果温度值在 10 秒钟之内连续上升,则报警。

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

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


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

env.execute();
}

// 自定义函数 检测一段时间(时间域)内温度连续上升,输出报警
private static class TempConsIncreWarring extends KeyedProcessFunction<Tuple, SensorReading, String> {
// 定义时间域
private Integer interval;

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

// 定义状态,保存上一个温度值,定时器时间戳
private ValueState<Double> lastTempState;
private ValueState<Long> tsTimeState;

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

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

// 如果温度上升就要注册10秒后的定时器且在无定时器时,等待
if (value.getTemperature() > lastTemp && tsTime == null) {
// 计算出定时器时间戳
Long ts = ctx.timerService().currentProcessingTime() + interval * 1000L;
// 注册定时器
ctx.timerService().registerProcessingTimeTimer(ts);
// 更新时间状态
tsTimeState.update(ts);
}
// 如果温度下降 需要删除定时器
if (value.getTemperature() < lastTemp && tsTime != null) {
ctx.timerService().deleteProcessingTimeTimer(tsTime);
// 清空时间状态定时器
tsTimeState.clear();
}

// 更新温度状态
lastTempState.update(value.getTemperature());
}

@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
// 定时器触发 输出报局信息
out.collect("传感器"+ ctx.getCurrentKey().getField(0) + "温度值连续"+ interval +"秒一直处于上升");
tsTimeState.clear();
}

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

运行结果:
结合案例总结Flink框架中的最底层API(ProcessFunction)用法_ide_02

侧流输出

侧流输出功能可以产生多条流,并且这些流的数据类型可以不一样。一个侧流可以定义为 OutputTag[X]对象,X 是输出流的数据类型。

一个案例:

监控传感器温度值,将温度值低于 30 度的数据输出到 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("localhost", 7777);

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


// 定义outputTag 表示低温流
OutputTag lowTemp = new OutputTag<SensorReading>("lowTemp") {
};

// 自定义测输出流实现分流操作
SingleOutputStreamOperator<SensorReading> highTempStream = dataStream.process(new ProcessFunction<SensorReading, SensorReading>() {

@Override
public void processElement(SensorReading value, Context ctx, Collector<SensorReading> out) throws Exception {
// 判断温度大于30 为高温流 输出到主流 低温流输出在侧流
if (value.getTemperature() > 30) {
out.collect(value);
} else {
ctx.output(lowTemp, value);
}
}
});

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


env.execute();
}
}

运行结果:
结合案例总结Flink框架中的最底层API(ProcessFunction)用法_flink_03


举报

相关推荐

0 条评论