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

百思不得小赵 2023-03-28 原文

概述

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

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 的 evenlong currentWatermark()
t time 定时器。当水位线大于等于定时器注册的时间时，触发定时器执行回调函数。

void registerEventTimeTimer(long timestamp)

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

voi 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();
        }
    }
}

运行结果：

侧流输出

侧流输出功能可以产生多条流，并且这些流的数据类型可以不一样。一个侧流可以定义为 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)用法的更多相关文章

ruby - 如何从 ruby 中的字符串运行任意对象方法？ - 2
总的来说，我对ruby还比较陌生，我正在为我正在创建的对象编写一些rspec测试用例。许多测试用例都非常基础，我只是想确保正确填充和返回值。我想知道是否有办法使用循环结构来执行此操作。不必为我要测试的每个方法都设置一个assertEquals。例如:describeitem,"TestingtheItem"doit"willhaveanullvaluetostart"doitem=Item.new#HereIcoulddotheitem.name.shouldbe_nil#thenIcoulddoitem.category.shouldbe_nilendend但我想要一些方法来使用
ruby - 其他文件中的 Rake 任务 - 2
我试图在一个项目中使用rake，如果我把所有东西都放到Rakefile中，它会很大并且很难读取/找到东西，所以我试着将每个命名空间放在lib/rake中它自己的文件中，我添加了这个到我的rake文件的顶部:Dir['#{File.dirname(__FILE__)}/lib/rake/*.rake'].map{|f|requiref}它加载文件没问题，但没有任务。我现在只有一个.rake文件作为测试，名为“servers.rake”，它看起来像这样:namespace:serverdotask:testdoputs"test"endend所以当我运行rakeserver:testid时
ruby-on-rails - Ruby net/ldap 模块中的内存泄漏 - 2
作为我的Rails应用程序的一部分，我编写了一个小导入程序，它从我们的LDAP系统中吸取数据并将其塞入一个用户表中。不幸的是，与LDAP相关的代码在遍历我们的32K用户时泄漏了大量内存，我一直无法弄清楚如何解决这个问题。这个问题似乎在某种程度上与LDAP库有关，因为当我删除对LDAP内容的调用时，内存使用情况会很好地稳定下来。此外，不断增加的对象是Net::BER::BerIdentifiedString和Net::BER::BerIdentifiedArray，它们都是LDAP库的一部分。当我运行导入时，内存使用量最终达到超过1GB的峰值。如果问题存在，我需要找到一些方法来更正我的代
ruby-on-rails - Rails 3 中的多个路由文件 - 2
Rails2.3可以选择随时使用RouteSet#add_configuration_file添加更多路由。是否可以在Rails3项目中做同样的事情？最佳答案在config/application.rb中:config.paths.config.routes在Rails3.2(也可能是Rails3.1)中，使用:config.paths["config/routes"] 关于ruby-on-rails-Rails3中的多个路由文件，我们在StackOverflow上找到一个类似的问题
ruby-on-rails - 结合 meta_search 与 acts_as_taggable_on - 2
我在开发的Rails3网站的一些搜索功能上遇到了一个小问题。我有一个简单的Post模型，如下所示:classPost我正在使用acts_as_taggable_on来更轻松地向我的帖子添加标签。当我有一个标记为“rails”的帖子并执行以下操作时，一切正常:@posts=Post.tagged_with("rails")问题是，我还想搜索帖子的标题。当我有一篇标题为“Helloworld”并标记为“rails”的帖子时，我希望能够通过搜索“hello”或“rails”来找到这篇帖子。因此，我希望标题列的LIKE语句与acts_as_taggable_on提供的tagged_with方法
ruby-on-rails - Rails - 一个 View 中的多个模型 - 2
我需要从一个View访问多个模型。以前，我的links_controller仅用于提供以不同方式排序的链接资源。现在我想包括一个部分(我假设)显示按分数排序的顶级用户(@users=User.all.sort_by(&:score))我知道我可以将此代码插入每个链接操作并从View访问它，但这似乎不是“ruby方式”，我将需要在不久的将来访问更多模型。这可能会变得很脏，是否有针对这种情况的任何技术？注意事项:我认为我的应用程序正朝着单一格式和动态页面内容的方向发展，本质上是一个典型的网络应用程序。我知道before_filter但考虑到我希望应用程序进入的方向，这似乎很麻烦。最终从任何
ruby-on-rails - Rails 3.2.1 中 ActionMailer 中的未定义方法 'default_content_type=' - 2
我在我的项目中添加了一个系统来重置用户密码并通过电子邮件将密码发送给他，以防他忘记密码。昨天它运行良好(当我实现它时)。当我今天尝试启动服务器时，出现以下错误。=>BootingWEBrick=>Rails3.2.1applicationstartingindevelopmentonhttp://0.0.0.0:3000=>Callwith-dtodetach=>Ctrl-CtoshutdownserverExiting/Users/vinayshenoy/.rvm/gems/ruby-1.9.3-p0/gems/actionmailer-3.2.1/lib/action_mailer
ruby-on-rails - Rails 应用程序中的 Rails : How are you using application_controller. rb 是新手吗？ - 2
刚入门rails，开始慢慢理解。有人可以解释或给我一些关于在application_controller中编码的好处或时间和原因的想法吗？有哪些用例。您如何为Rails应用程序使用应用程序Controller？我不想在那里放太多代码，因为据我了解，每个请求都会调用此Controller。这是真的？最佳答案 ApplicationController实际上是您应用程序中的每个其他Controller都将从中继承的类(尽管这不是强制性的)。我同意不要用太多代码弄乱它并保持干净整洁的态度，尽管在某些情况下ApplicationContr
ruby-on-rails - form_for 中不在模型中的自定义字段 - 2
我想向我的Controller传递一个参数，它是一个简单的复选框，但我不知道如何在模型的form_for中引入它，这是我的观点:{:id=>'go_finance'}do|f|%>Transferirde:para:Entrada:"input",:placeholder=>"Quantofoiganho?"%>Saída:"output",:placeholder=>"Quantofoigasto?"%>Nota:我想做一个额外的复选框，但我该怎么做，模型中没有一个对象，而是一个要检查的对象，以便在Controller中创建一个ifelse，如果没有检查，请帮助我，非常感谢,谢谢
ruby - rspec 需要 .rspec 文件中的 spec_helper - 2
我注意到像bundler这样的项目在每个specfile中执行requirespec_helper我还注意到rspec使用选项--require，它允许您在引导rspec时要求一个文件。您还可以将其添加到.rspec文件中，因此只要您运行不带参数的rspec就会添加它。使用上述方法有什么缺点可以解释为什么像bundler这样的项目选择在每个规范文件中都需要spec_helper吗？最佳答案我不在Bundler上工作，所以我不能直接谈论他们的做法。并非所有项目都checkin.rspec文件。原因是这个文件，通常按照当前的惯例，只

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

概述

KeyedProcessFunction<K, I, O>

定时器

侧流输出

有关结合案例总结Flink框架中的最底层API（ProcessFunction)用法的更多相关文章

随机推荐