大数据-玩转数据-Flink状态编程（中）

news2024/11/25 4:19:12

一、键控状态

键控状态是根据输入数据流中定义的键（key）来维护和访问的。
Flink为每个键值维护一个状态实例，并将具有相同键的所有数据，都分区到同一个算子任务中，这个任务会维护和处理这个key对应的状态。当任务处理一条数据时，它会自动将状态的访问范围限定为当前数据的key。因此，具有相同key的所有数据都会访问相同的状态。
Keyed State很类似于一个分布式的key-value map数据结构，只能用于KeyedStream（keyBy算子处理之后）。

二、ValueState

保存单个值. 每个key有一个状态值. 设置使用 update(T), 获取使用 T value()

案例：
检测传感器的水位值，如果连续的两个水位值超过10，就输出报警

package com.lyh.flink09;

import com.lyh.bean.WaterSensor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class key_Value {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.socketTextStream("hadoop100",9999)
                .map(line -> {
                    String[] datas = line.split(",");
                    return  new WaterSensor(
                            datas[0],
                            Long.valueOf(datas[1]),
                            Integer.valueOf(datas[2]));

                })
                .keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, String>() {

                    private ValueState<Integer> lastValueState;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        lastValueState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("ValueState",Integer.class));
                    }

                    @Override
                    public void processElement(WaterSensor value,
                                               Context ctx,
                                               Collector<String> out) throws Exception {
                        Integer lastVc = lastValueState.value();
                        if (lastVc != null) {
                            if (lastVc > 10 && value.getVc()>10) {
                                out.collect(ctx.getCurrentKey() + "连续两次超越告警值。。。。");

                            }

                        }
                        lastValueState.update(value.getVc());
                    }

                }).print();
        env.execute();

    }
}

三、ListState

针对每个传感器输出最高的3个水位值

package com.lyh.flink09;

import com.lyh.bean.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

public class key_List {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        env.socketTextStream("hadoop100",9999)
                .map(line ->{
                    String[] datas = line.split(",");
                    return new WaterSensor(
                            datas[0],
                            Long.valueOf(datas[1]),
                            Integer.valueOf(datas[2])
                    );
                }).keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, List<Integer>>() {
                    private ListState<Integer> top3Vc;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        top3Vc = getRuntimeContext().getListState(new ListStateDescriptor<Integer>("top3Vc", Integer.class));
                    }
                    @Override
                    public void processElement(WaterSensor value,
                                               Context ctx,
                                               Collector<List<Integer>> out) throws Exception {
                       top3Vc.add(value.getVc());
                        Iterable<Integer> it = top3Vc.get();
                        List<Integer> vcs = new ArrayList<>();
                        for (Integer vc : top3Vc.get()) {
                            vcs.add(vc);
                        }
                        vcs.sort((o1,o2)->o2-o1);
                        if (vcs.size()>3) {
                            vcs.remove(3);
                        }
                        top3Vc.update(vcs);
                        out.collect(vcs);
                    }
                }).print();
        env.execute();
    }

}

运行结果：

2> [11]
2> [11, 1]
2> [11, 1, 1]
2> [100, 11, 1]
2> [1000, 100, 11]
2> [1000, 100, 99]
2> [1000, 100, 99]

四、ReducingState

计算每个传感器的水位和

package com.lyh.flink09;

import com.lyh.bean.WaterSensor;
import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class key_Reduce {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        env.socketTextStream("hadoop100",9999)
                .map(line -> {
                    String[] datas = line.split(",");
                    return new WaterSensor(
                    datas[0],
                    Long.valueOf(datas[1]),
                    Integer.valueOf(datas[2]));
                }).keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, Integer>() {
                    private ReducingState<Integer> vcstate;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                         vcstate = getRuntimeContext().getReducingState(new ReducingStateDescriptor<Integer>(
                                "vcstate",
                                Integer::sum,
                                Integer.class));
                    }
                    @Override
                    public void processElement(WaterSensor value,
                                               Context ctx,
                                               Collector<Integer> out) throws Exception {
                              vcstate.add(value.getVc());
                              out.collect(vcstate.get());

                    }


                }).print();
        env.execute();
    }
}

运行：
在这里插入图片描述
运行结果：
在这里插入图片描述

五、AggregatingState

计算每个传感器的平均水位

package com.lyh.flink09;

import com.lyh.bean.WaterSensor;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.AggregatingState;
import org.apache.flink.api.common.state.AggregatingStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class key_Agg {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        env.socketTextStream("hadoop100",9999)
                .map(line ->{
                    String[] datas = line.split(",");
                    return new WaterSensor(
                            datas[0],
                            Long.valueOf(datas[1]),
                            Integer.valueOf(datas[2])
                    );
                }).keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, Double>() {
                    private AggregatingState<Integer, Double> avgState;
                    @Override
                    public void open(Configuration parameters) throws Exception {

                        AggregatingStateDescriptor<Integer, Tuple2<Integer, Integer>, Double>
                         Avg = new AggregatingStateDescriptor<>(
                                "Avg",
                                new AggregateFunction<Integer,
                                        Tuple2<Integer, Integer>,
                                        Double>() {
                            @Override
                            public Tuple2<Integer, Integer> createAccumulator() {
                                return Tuple2.of(0,0);
                            }

                            @Override
                            public Tuple2<Integer, Integer> add(Integer value, Tuple2<Integer, Integer> integerTuple2) {
                                return Tuple2.of(integerTuple2.f0 + value,integerTuple2.f1 + 1);
                            }

                            @Override
                            public Double getResult(Tuple2<Integer, Integer> integerTuple2) {
                                return integerTuple2.f0*1D/integerTuple2.f1;
                            }

                            @Override
                            public Tuple2<Integer, Integer> merge(Tuple2<Integer, Integer> a, Tuple2<Integer, Integer> b) {
                                return Tuple2.of(a.f0 + b.f0 , a.f1 + b.f1);
                            }
                        }, Types.TUPLE(Types.INT,Types.INT));
                        avgState = getRuntimeContext().getAggregatingState(Avg);
                    }

                    @Override
                    public void processElement(WaterSensor value,
                                               Context ctx,
                                               Collector<Double> out) throws Exception {
                        avgState.add(value.getVc());
                        out.collect(avgState.get());

                    }
                }).print();
        env.execute();
    }
}

运行：
在这里插入图片描述
结果：
在这里插入图片描述

六、MapState

去重: 去掉重复的水位值. 思路: 把水位值作为MapState的key来实现去重, value随意

package com.lyh.flink09;

import com.lyh.bean.WaterSensor;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class key_Map {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        env.socketTextStream("hadoop100",9999)
                .map(line ->{
                    String[] datas = line.split(",");
                    return new WaterSensor (
                        datas[0],
                        Long.valueOf(datas[1]),
                        Integer.valueOf(datas[2]));
                }).keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, WaterSensor>() {

                    private MapState<Integer, String> mapState;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                         mapState = this.getRuntimeContext().getMapState(new MapStateDescriptor<Integer, String>("mapState", Integer.class, String.class));

                    }

                    @Override
                    public void processElement(WaterSensor values,
                                               Context ctx,
                                               Collector<WaterSensor> out) throws Exception {
                        if (!mapState.contains(values.getVc())) {
                            out.collect(values);
                            mapState.put(values.getVc(),"随意");

                        }
                    }
                }).print();
        env.execute();
    }
}

运行：
在这里插入图片描述
结果：
在这里插入图片描述

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/994827.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

Jmeter压测监控体系搭建Docker+Influxdb+Grafana

Jmeter压测监控体系搭建Docker+Influxdb+Grafana

章节目录&#xff1a; 一、背景介绍1.1 概述1.2 拓扑图 二、云服务器设置三、Docker3.1 概述3.2 搭建流程3.3 安装验证3.4 配置docker镜像加速3.5 取消sudo运行(可选操作) 四、InfluxDB4.1 镜像拉取4.2 运行数据库4.3 创建存储 jmeter 数据的库 五、Grafana5.1 镜像拉取5.2 关联…
阅读更多...
Day_13 ＞ 指针进阶（2）

Day_13 ＞ 指针进阶（2）

目录 1.函数指针数组 2.指向函数指针数组的指针 3.回调函数 qsort()函数 代码示例 void* 4.结束 今天我们在进阶指针的基础上&#xff0c;学习进阶指针的第二部分 1.函数指针数组 首先我们回顾一下指针数组 char* arr[5]://字符指针数组 - 数组 - 存放的是字符指针 in…
阅读更多...
mysql的索引结构

mysql的索引结构

索引概述 索引&#xff08; index &#xff09;是帮助 MySQL 高效获取数据的数据结构 ( 有序 ) 。在数据之外&#xff0c;数据库系统还维护着满足特定查找算法的数据结构&#xff0c;这些数据结构以某种方式引用&#xff08;指向&#xff09;数据&#xff0c; 这样就可以在这些…
阅读更多...
Spring与OAuth2：实现第三方认证和授权的最佳实践

Spring与OAuth2：实现第三方认证和授权的最佳实践

&#x1f337;&#x1f341; 博主猫头虎&#xff08;&#x1f405;&#x1f43e;&#xff09;带您 Go to New World✨&#x1f341; &#x1f984; 博客首页——&#x1f405;&#x1f43e;猫头虎的博客&#x1f390; &#x1f433; 《面试题大全专栏》 &#x1f995; 文章图文…
阅读更多...
python基本类型

python基本类型

数值类型 整型 int_val 1145143 print(int_val)python中的整型是大数类型。 一些其他函数 val 30 vlen val.bit_length() # 转换为二进制的长度 v_8 oct(val) print(v_8) # 将十进制转为八进制 v_16 hex(val) # 将十进制转为十六进制 v_2 bin(val) # 将十进制转为二进…
阅读更多...
二、环境配置，项目运行 —— TinyWebServer

二、环境配置，项目运行 —— TinyWebServer

环境配置&#xff0c;项目运行 —— TinyWebServer 一、前言 上一期已经介绍过这个项目的基本结构&#xff0c;不懂得可以点开主页查找。 写代码前。一般的步骤就是&#xff0c;先把别人的代码下载下来运行。一、一方面看看最终效果是否是自己想要的&#xff0c;二、掌握项目…
阅读更多...
redis分布式锁详解

redis分布式锁详解

一、基本分布式锁实现 1、案例&#xff08;扣减库存&#xff09; RequestMapping("reduceStock")public String reduceStock() {String lockKey "lock:product_101";String clientId UUID.randomUUID().toString();// 过期时间要和设置key成为一条命令…
阅读更多...
linux下shell脚本实现wordpress搭建

linux下shell脚本实现wordpress搭建

wordpress_auto_install.sh #!/bin/bashuser$(whoami)function wordpress_auto_install () { if [ $user "root" ];thenecho "前提&#xff1a;调整系统配置&#xff0c;如关闭selinux、firewall等&#xff01;"sed -i s/SELINUXenforcing/SELINUXdis…
阅读更多...
光线投射之伪3d

光线投射之伪3d

光线投射是一种在 2D 地图中创建 3D 透视的渲染技术。当计算机速度较慢时&#xff0c;不可能实时运行真正的 3D 引擎&#xff0c;光线投射是第一个解决方案。光线投射可以非常快&#xff0c;因为只需对屏幕的每条垂直线进行计算。 光线投射的基本思想如下&#xff1a;地图是一…
阅读更多...
rtthread下基于spi device架构MCP25625驱动

rtthread下基于spi device架构MCP25625驱动

1.CAN驱动架构 由于采用了RTT的spi device架构&#xff0c;不能再随心所遇的编写CAN驱动 了&#xff0c;之前内核虽然采用了RTT内核&#xff0c;但是驱动并没有严格严格按RTT推荐的架构来做&#xff0c;这次不同了&#xff0c;上次是因为4个MCP25625挂在了4路独立的SPI总线上&…
阅读更多...
【图论】Floyd

【图论】Floyd

算法提高课笔记&#xff09; 文章目录 例题牛的旅行题意思路代码 排序题意思路代码 观光之旅题意思路代码 例题 牛的旅行 原题链接 农民John的农场里有很多牧区&#xff0c;有的路径连接一些特定的牧区。 一片所有连通的牧区称为一个牧场。 但是就目前而言&#xff0c;你…
阅读更多...
程序依赖相关知识点(PDG,SDG)

程序依赖相关知识点(PDG,SDG)

什么叫可达性 变量v的定义d&#xff1a;对变量v的赋值语句称为变量v的定义 变量v的使用&#xff1a;在某个表达式中引用变量v的值 当变量v被再次赋值时&#xff0c;上一次赋值对变量v的定义d就被kill掉了 如果定义d到点p之间存在一条路径&#xff0c;且在路径中定义d没有被…
阅读更多...
Java 多线程系列Ⅵ（并发编程的五大组件）

Java 多线程系列Ⅵ（并发编程的五大组件）

JUC 组件 前言一、Callable二、ReentrantLock三、Atomic 原子类四、线程池五、Semaphore六、CountDownLatch 前言 JUC&#xff08;Java.util.concurrent&#xff09;是 Java 标准库中的一个包&#xff0c;它提供了一组并发编程工具&#xff0c;本篇文章就介绍几组常见的 JUC 组…
阅读更多...
汇川PLC学习Day2：编写检测IO端口状态程序

汇川PLC学习Day2：编写检测IO端口状态程序

汇川PLC学习Day2&#xff1a;编写检测IO端口状态程序 一、 新增IO和模拟量模块 IO组态界面 模块参数设置 程序编写 想法是将DA模块的通道0接到AD模块的通道0&#xff0c;将DA模块的通道1接到AD模块的通道1&#xff0c;PLC本身发模拟量给自己PLC收模拟量转换&#xff0c;…
阅读更多...
MySQL 8.0.25版本下载、安装及配置（Windows 10/11 64位）详细教程【超详细，保姆级教程！！！】

MySQL 8.0.25版本下载、安装及配置（Windows 10/11 64位）详细教程【超详细，保姆级教程！！！】

本文介绍关于windows 11如何安装配置MySQL 8.0.25版本的详细步骤 MySQL下载地址&#xff08;官网&#xff09; 一、下载MySQL 8.0.25 1、进入官网&#xff0c;选择版本 8.0.25 2、下载MySQL压缩包 3、下载完成后将压缩包解压至方便自己查找的位置&#xff08;切记&#xf…
阅读更多...
Tensor数据转换为稀疏矩阵

Tensor数据转换为稀疏矩阵

Tensor数据转换为稀疏矩阵 一、稀疏矩阵 原文链接 常用的稀疏矩阵存储格式有COO&#xff0c;CSR/CSC&#xff0c;LIL 1.COO COO(Coordinate format )是最为简单的格式&#xff0c;以三元组的形式存储稀疏矩阵。记录矩阵中非零元素的数值和所在的行序号和列序号。形式为&am…
阅读更多...
工商业储能CE认证电表ADW300

工商业储能CE认证电表ADW300

安科瑞 华楠 ADW300 无线计量仪表主要用于计量低压网络的三相有功电能&#xff0c;具有体积小、精度高、功能丰富等优点&#xff0c;并且可选通讯方式多&#xff0c;可支持 RS485 通讯和 Lora、NB、4G、wifi 等无线通讯方式&#xff0c;增加了外置互感器的电流采样模式&#x…
阅读更多...
【数据结构】线性表

【数据结构】线性表

线性表 顺序表链式存储单链表双链表 知识目录 顺序表 概念&#xff1a;用一组地址连续的存储单元依次存储线性表的数据元素&#xff0c;这种存储结构的线性表称为顺序表。 特点&#xff1a;逻辑上相邻的数据元素&#xff0c;物理次序也是相邻的。 只要确定好了存储线性表的…
阅读更多...
基本数据类型和包装类型 使用规范

基本数据类型和包装类型 使用规范

使用规范 1 概念1.1 基本数据类型1.2 包装类型1.3 对应关系1.4 自动装箱/拆箱 2 变量类型2.1 全局变量2.1.1 常量&#xff08;Constants&#xff09;2.1.2 类变量&#xff08;Class Variables&#xff09;2.1.3 实例变量&#xff08;Instance Variables&#xff09; 2.2 局部变…
阅读更多...
快速实现抖音上下滑动，你不知道的ViewPager2用法，信息量巨大，建议收藏点赞。老tier~

快速实现抖音上下滑动，你不知道的ViewPager2用法，信息量巨大，建议收藏点赞。老tier~

万能ViewPager2适配器–SmartViewPager2Adapter 特点功能 完全脱离xml&#xff0c;所有效果只需要通过api调用 具体功能&#xff1a;1. 两句代码实现抖音列表效果2. 无感且丝滑&#xff0c;动态从头部或者底部加载数据3. 设置上下加载监听&#xff0c;再达到预加载limit的时…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023