flink自定义process,使用状态求历史总和(scala)
es idea maven 依赖
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-elasticsearch7_2.11</artifactId>
<version>1.11.1</version>
</dependency>
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.java.typeutils.TypeExtractor
import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.elasticsearch.{ElasticsearchSinkFunction, RequestIndexer}
import org.apache.flink.streaming.connectors.elasticsearch7.{ElasticsearchSink, RestClientFactory}
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer, KafkaDeserializationSchema}
import org.apache.flink.util.Collector
import org.apache.http.HttpHost
import org.apache.http.client.config.RequestConfig
import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.elasticsearch.action.DocWriteRequest
import org.elasticsearch.client.{Requests, RestClientBuilder}import java.time.Duration
import java.util.Propertiesobject Test {def main(args: Array[String]): Unit = {val env = StreamExecutionEnvironment.getExecutionEnvironment//需要状态开启下面的配置//env.setStateBackend(new RocksDBStateBackend(s"hdfs://${namenodeID}", true))//hdfs 作为状态后端//env.enableCheckpointing(10 * 60 * 1000L)//env.getCheckpointConfig.setCheckpointTimeout(10 * 60 * 1000L)env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) //事件时间val props = new Propertiesprops.setProperty("bootstrap.servers", "host:6667") //有些是9092端口props.setProperty("group.id", "groupId")props.setProperty("retries", "10")props.setProperty("retries.backoff.ms", "100")props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, "60000")//是否配置了权限,有的话加上下面的配置// props.setProperty("sasl.jaas.config","org.apache.kafka.common.security.plain.PlainLoginModule required username='' password='';")//props.setProperty("security.protocol", "SASL_PLAINTEXT");// props.setProperty("sasl.mechanism", "PLAIN")val httpHosts = new java.util.ArrayList[HttpHost]httpHosts.add(new HttpHost("esIPOne", 9200, "http"))httpHosts.add(new HttpHost("esIPTwo", 9200, "http"))httpHosts.add(new HttpHost("esIPThree", 9200, "http"))val esSinkBuilder = new ElasticsearchSink.Builder[ResultBean](httpHosts, new ElasticsearchSinkFunction[ResultBean] {def process(element: ResultBean, ctx: RuntimeContext, indexer: RequestIndexer) {val json = new java.util.HashMap[String, Any]json.put("@timestamp", element.ts)json.put("data", element.data)json.put("sum", element.sum)val rqst = Requests.indexRequest().index("indexName").id(element.id).source(json).opType(DocWriteRequest.OpType.INDEX)indexer.add(rqst)}})setESConf(esSinkBuilder, 5000)val myConsumer = new FlinkKafkaConsumer[DemoBean]("topicName", new DemoKafka(), props).setStartFromEarliest() //从什么时间开始读val source = env.addSource(myConsumer).uid("source-data").name("数据源").assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness[DemoBean](Duration.ofSeconds(1)).withTimestampAssigner(new SerializableTimestampAssigner[DemoBean] {override def extractTimestamp(element: DemoBean, recordTimestamp: Long): Long = element.ts}).withIdleness(Duration.ofSeconds(5))).uid("water-marks").name("注册水位线")source.keyBy(k => k.id).process(new DemoProcess()).uid("demo-process").name("process 示例").addSink(esSinkBuilder.build()).uid("es-sink").name("数据写入es")env.execute("任务名")}private class DemoKafka() extends KafkaDeserializationSchema[DemoBean] {override def isEndOfStream(t: DemoBean): Boolean = falseoverride def deserialize(consumerRecord: ConsumerRecord[Array[Byte], Array[Byte]]): DemoBean = {val value = new String(consumerRecord.value())val list = value.split("\t")DemoBean(list(0), list(1), list(2).toInt, list(3).toLong)}override def getProducedType: TypeInformation[DemoBean] = TypeExtractor.getForClass(classOf[DemoBean])}private class DemoProcess extends KeyedProcessFunction[String, DemoBean, ResultBean] {private var hisSumState: ValueState[Int] = _override def open(parameters: Configuration): Unit = {hisSumState = getRuntimeContext.getState(new ValueStateDescriptor("his-sum", classOf[Int]))}override def processElement(data: DemoBean, ctx: KeyedProcessFunction[String, DemoBean, ResultBean]#Context, out: Collector[ResultBean]): Unit = {val his = if (hisSumState.value() == null) 0 else hisSumState.value()val now = data.valuehisSumState.update(now)out.collect(ResultBean(data.id, data.data, his + now, data.value))}}def setESConf[T](esSinkBuilder: ElasticsearchSink.Builder[T], numMaxActions: Int) {esSinkBuilder.setBulkFlushMaxActions(numMaxActions)esSinkBuilder.setBulkFlushMaxSizeMb(10)esSinkBuilder.setBulkFlushInterval(10000)esSinkBuilder.setBulkFlushBackoff(true)esSinkBuilder.setBulkFlushBackoffDelay(2)esSinkBuilder.setBulkFlushBackoffRetries(3)esSinkBuilder.setRestClientFactory(new RestClientFactory {override def configureRestClientBuilder(restClientBuilder: RestClientBuilder): Unit = {restClientBuilder.setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {override def customizeRequestConfig(requestConfigBuilder: RequestConfig.Builder): RequestConfig.Builder = {requestConfigBuilder.setConnectTimeout(12000)requestConfigBuilder.setSocketTimeout(90000)}})}})}private case class DemoBean(id: String, data: String, value: Int, ts: Long)private case class ResultBean(id: String, data: String, sum: Int, ts: Long)
}
相关文章:
flink自定义process,使用状态求历史总和(scala)
es idea maven 依赖 <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-elasticsearch7_2.11</artifactId> <version>1.11.1</version> </dependency> import org.apache.flink.api.common.eve…...
股指期货理论价格计算公式是什么?
股指期货,作为金融衍生品的一种,其价格与现货市场的股指价格紧密相关,但又受到多种因素的影响。了解股指期货理论价格的计算公式,对于投资者进行套利交易、风险管理等具有重要意义。本文将详细解读股指期货理论价格的计算公式&…...
解决R包依赖版本不兼容问题
ERROR: dependency ‘Matrix’ is not available for package ‘irlba’ removing ‘/root/anaconda3/envs/myview/lib/R/library/irlba’ ERROR: dependency ‘Matrix’ is not available for package ‘N2R’ removing ‘/root/anaconda3/envs/myview/lib/R/library/N2R’ ER…...
HarmonyOS开发者基础认证考试试题
文章目录 一、判断题二、单选题三、多选题 因考试只有91分,所以下方答案有部分错误,如果有发现错误,欢迎提出 一、判断题 1. HarmonyOS提供了基础的应用加固安全能力,包括混淆、加密和代码签名能力 正确 2. 用户首选项是关系型数…...
如何使用 React、TypeScript、TailwindCSS 和 Vite 创建 Chrome 插件
创建一个 Chrome 插件是一个有趣的项目,特别是当结合使用强大的工具如 React、TypeScript、TailwindCSS 和 Vite 时 在这篇文章中,我们将逐步引导完成整个过程,了解如何在 2024 年构建自己的 Chrome 插件。无论是经验丰富的开发者还是刚刚起…...
机器学习——Stacking
Stacking: 方法:训练多个模型(可以是强模型),然后将这些模型的预测结果作为新的特征,输入到下一层新的模型(可以是多个)中进行训练,从而得到最终的预测结果。 代表:Stacking本身并没…...
在HTML中添加图片
在HTML中添加图片,你需要使用<img>标签。这个标签用于在网页上嵌入图像。<img>是一个空元素,它只包含属性,并且没有闭合标签。要在<img>标签中指定要显示的图像,你需要使用src(source的缩写…...
R语言机器学习算法实战系列(二) SVM算法(Support Vector Machine)
文章目录 介绍原理应用方向下载数据加载R包导入数据数据预处理数据描述数据切割标准化数据设置参数训练模型预测测试数据评估模型模型准确性混淆矩阵模型评估指标ROC CurvePRC Curve特征的重要性保存模型总结系统信息介绍 支持向量机(Support Vector Machine,简称SVM)是一种…...
gdb调试使用记录
使用 GDB(GNU Debugger)进行问题排查是非常有效的。且可以通过core文件进行排查bug,core文件是程序异常崩溃的时候(段错误,非法指令等),系统自动生成的core文件。用户可以通过core文件配合gdb调试命令,调试…...
ESXi安装【真机和虚拟机】(超详细)
项目简介: ESXi(Elastic Sky X Integrated)是VMware公司开发的一种裸机虚拟化管理程序,允许用户在单一物理服务器上运行多个虚拟机(VM)。它直接安装在服务器硬件上,而不是操作系统之上ÿ…...
基于SpringBoot+Vue的高校门禁管理系统
作者:计算机学姐 开发技术:SpringBoot、SSM、Vue、MySQL、JSP、ElementUI、Python、小程序等,“文末源码”。 专栏推荐:前后端分离项目源码、SpringBoot项目源码、SSM项目源码 精品专栏:Java精选实战项目源码、Python精…...
【Linux-基础IO】C语言文件接口回顾 系统文件概念及接口
目录 一、C语言文件接口回顾 C语言基础知识 C中文件操作示例 二、系统文件概念及接口 重定向基本理解的回顾 文件的基本概念 系统调用接口 open read write close lseek 什么是当前路径 一、C语言文件接口回顾 引言:我们并不理解文件!从语…...
系统架构笔记-3-信息系统基础知识
知识要点 结构化方法:结构是指系统内各个组成要素之间的相互联系、相互作用的框架。结构化方法也称为生命周期法,是一种传统的信息系统开发方法,由结构化分析、结构化设计、结构化程序设计三部分有机组合而成,精髓是自顶向下、逐…...
Linux下编程实现网络传送文件
本程序是在Linux下开发的,使用的是C语言,再结合Socket进行编程,分为客户端和服务器两个程序,即采用的是C/S架构,相应的源代码如下: 服务器端: #include <stdio.h> //#include <stdlib.h> #include <sys/socket.h> #include <netinet/in.h&g…...
【速成Redis】04 Redis 概念扫盲:事务、持久化、主从复制、哨兵模式
前言: 前三篇如下: 【速成Redis】01 Redis简介及windows上如何安装redis-CSDN博客 【速成Redis】02 Redis 五大基本数据类型常用命令-CSDN博客 【速成Redis】03 Redis 五大高级数据结构介绍及其常用命令 | 消息队列、地理空间、HyperLogLog、BitMap、…...
SQL Server 2022的数据类型
新书速览|SQL Server 2022从入门到精通:视频教学超值版_sql server 2022 出版社-CSDN博客 《SQL Server 2022从入门到精通(视频教学超值版)(数据库技术丛书)》(王英英)【摘要 书评 试读】- 京东图书 (jd.com) 数据类…...
Linux基础3-基础工具4(git),冯诺依曼计算机体系结构
上篇文章:Linux基础3-基础工具3(make,makefile,gdb详解)-CSDN博客 本章重点: 1. git简易使用 2. 冯诺依曼计算机体系结构介绍 目录 一. git使用 1.1 什么是git? 1.2 git发展史 1.3 git创建仓库 1.4 git命令操作 二. 冯诺依…...
后台数据管理系统 - 项目架构设计-Vue3+axios+Element-plus(0916)
接口文档: https://apifox.com/apidoc/shared-26c67aee-0233-4d23-aab7-08448fdf95ff/api-93850835 接口根路径: http://big-event-vue-api-t.itheima.net 本项目的技术栈 本项目技术栈基于 ES6、vue3、pinia、vue-router 、vite 、axios 和 element-plus http:/…...
MySQL基础篇(黑马程序员2022-01-18)
1 MySQL数据库概述 1.1 MySQL数据库的下载,安装,启动停止 1.2 数据模型 (1)关系型数据库(RDBMS) 概念:建立在关系模型基础上,由多张相互连接的二维表组成的数据库。 特点: A. 使用表存储数据,格式统一,便于维护。…...
nodejs 013:Prect 样式复用(multiple classes)例子
Prect 简单示例 Prect 为使用相同的现代 API 的快速 3kB React 替代方案。代码形式与 React 基本相同。部分语法区别可见 prect-differences-to-react。以下是一个 Prect 简单示例。 Button目录Button.css: .this {display: inline-block;padding: 3px 8px;margi…...
SCAU期末笔记 - 数据分析与数据挖掘题库解析
这门怎么题库答案不全啊日 来简单学一下子来 一、选择题(可多选) 将原始数据进行集成、变换、维度规约、数值规约是在以下哪个步骤的任务?(C) A. 频繁模式挖掘 B.分类和预测 C.数据预处理 D.数据流挖掘 A. 频繁模式挖掘:专注于发现数据中…...
家政维修平台实战20:权限设计
目录 1 获取工人信息2 搭建工人入口3 权限判断总结 目前我们已经搭建好了基础的用户体系,主要是分成几个表,用户表我们是记录用户的基础信息,包括手机、昵称、头像。而工人和员工各有各的表。那么就有一个问题,不同的角色…...
跨链模式:多链互操作架构与性能扩展方案
跨链模式:多链互操作架构与性能扩展方案 ——构建下一代区块链互联网的技术基石 一、跨链架构的核心范式演进 1. 分层协议栈:模块化解耦设计 现代跨链系统采用分层协议栈实现灵活扩展(H2Cross架构): 适配层…...
NLP学习路线图(二十三):长短期记忆网络(LSTM)
在自然语言处理(NLP)领域,我们时刻面临着处理序列数据的核心挑战。无论是理解句子的结构、分析文本的情感,还是实现语言的翻译,都需要模型能够捕捉词语之间依时序产生的复杂依赖关系。传统的神经网络结构在处理这种序列依赖时显得力不从心,而循环神经网络(RNN) 曾被视为…...
C++八股 —— 单例模式
文章目录 1. 基本概念2. 设计要点3. 实现方式4. 详解懒汉模式 1. 基本概念 线程安全(Thread Safety) 线程安全是指在多线程环境下,某个函数、类或代码片段能够被多个线程同时调用时,仍能保证数据的一致性和逻辑的正确性…...
C# 求圆面积的程序(Program to find area of a circle)
给定半径r,求圆的面积。圆的面积应精确到小数点后5位。 例子: 输入:r 5 输出:78.53982 解释:由于面积 PI * r * r 3.14159265358979323846 * 5 * 5 78.53982,因为我们只保留小数点后 5 位数字。 输…...
逻辑回归暴力训练预测金融欺诈
简述 「使用逻辑回归暴力预测金融欺诈,并不断增加特征维度持续测试」的做法,体现了一种逐步建模与迭代验证的实验思路,在金融欺诈检测中非常有价值,本文作为一篇回顾性记录了早年间公司给某行做反欺诈预测用到的技术和思路。百度…...
Git常用命令完全指南:从入门到精通
Git常用命令完全指南:从入门到精通 一、基础配置命令 1. 用户信息配置 # 设置全局用户名 git config --global user.name "你的名字"# 设置全局邮箱 git config --global user.email "你的邮箱example.com"# 查看所有配置 git config --list…...
适应性Java用于现代 API:REST、GraphQL 和事件驱动
在快速发展的软件开发领域,REST、GraphQL 和事件驱动架构等新的 API 标准对于构建可扩展、高效的系统至关重要。Java 在现代 API 方面以其在企业应用中的稳定性而闻名,不断适应这些现代范式的需求。随着不断发展的生态系统,Java 在现代 API 方…...
WEB3全栈开发——面试专业技能点P4数据库
一、mysql2 原生驱动及其连接机制 概念介绍 mysql2 是 Node.js 环境中广泛使用的 MySQL 客户端库,基于 mysql 库改进而来,具有更好的性能、Promise 支持、流式查询、二进制数据处理能力等。 主要特点: 支持 Promise / async-await…...
