Flink SQL自定义标量函数(Scalar Function)
使用场景: 标量函数即 UDF,⽤于进⼀条数据出⼀条数据的场景。
开发流程:
- 实现 org.apache.flink.table.functions.ScalarFunction 接⼝
- 实现⼀个或者多个⾃定义的 eval 函数,名称必须叫做 eval,eval ⽅法签名必须是 public 的
- eval ⽅法的⼊参、出参都是直接体现在 eval 函数的签名中
开发案例:
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.InputGroup;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;
import static org.apache.flink.table.api.Expressions.*;/*** 输入数据: * nc -lk 88888* a,1** 输出结果:* res1=>:3> +I[97]* res2=>:3> +I[97]* res3=>:3> +I[97]*/
public class ScalarFunctionTest {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);DataStreamSource<String> source = env.socketTextStream("localhost", 8888);SingleOutputStreamOperator<Tuple2<String, String>> tpStream = source.map(new MapFunction<String, Tuple2<String, String>>() {@Overridepublic Tuple2<String, String> map(String input) throws Exception {return new Tuple2<>(input.split(",")[0], input.split(",")[1]);}});Table table = tEnv.fromDataStream(tpStream, "id,name");tEnv.createTemporaryView("SourceTable",table);// 在 Table API ⾥不经注册直接调⽤函数Table res1 = tEnv.from("SourceTable").select(call(HashFunction.class, $("id")));// 注册函数tEnv.createTemporarySystemFunction("HashFunction", HashFunction.class);// 在 Table API ⾥调⽤注册好的函数Table res2 = tEnv.from("SourceTable").select(call("HashFunction", $("id")));// 在 SQL ⾥调⽤注册好的函数Table res3 = tEnv.sqlQuery("SELECT HashFunction(id) FROM SourceTable");tEnv.toDataStream(res1).print("res1=>");tEnv.toDataStream(res2).print("res2=>");tEnv.toDataStream(res3).print("res3=>");env.execute();}public static class HashFunction extends ScalarFunction {// 接受任意类型输⼊,返回 INT 型输出public int eval(@DataTypeHint(inputGroup = InputGroup.ANY) Object o) {return o.hashCode();}}
}
测试结果:

相关文章:
Flink SQL自定义标量函数(Scalar Function)
使用场景: 标量函数即 UDF,⽤于进⼀条数据出⼀条数据的场景。 开发流程: 实现 org.apache.flink.table.functions.ScalarFunction 接⼝实现⼀个或者多个⾃定义的 eval 函数,名称必须叫做 eval,eval ⽅法签名必须是 p…...
【第2章 Node.js基础】2.4 Node.js 全局对象(二) process 对象
process对象是一个全局对象,提供当前Node.js 进程信息并对其进行控制。通常用于编写本地命令行程序。 1.进程事件 process对象是EventEmitter类的实例,因此可以使用事件的方式来处理和监听process对象的各种事件。以下是一些常用的process对象事件&…...
觉非科技发布【轻地图高速NOA智驾方案】|地平线,觉非科技,MobileDrive超捷生态协作实现技术落地
11月10日,2023世界新汽车技术合作生态展期间,觉非科技正式发布了基于地平线征程系列以及MobileDrive超捷规控系统的【轻地图高速NOA智驾解决方案】。该方案首次实现了从芯片能力输出、到数据闭环建设、再到规控部署的产业生态链协作与打通,为…...
竞赛 车道线检测(自动驾驶 机器视觉)
0 前言 无人驾驶技术是机器学习为主的一门前沿领域,在无人驾驶领域中机器学习的各种算法随处可见,今天学长给大家介绍无人驾驶技术中的车道线检测。 1 车道线检测 在无人驾驶领域每一个任务都是相当复杂,看上去无从下手。那么面对这样极其…...
128. 最长连续序列
这道题最简单的想法就是排序计数,但是复杂度为O(nlogn),不符合题意 于是采用哈希表的方法 将所有数字存放在哈希表中,然后开始逐个寻找。 比如当前遍历到x,如果x-1也存在哈希表中,那就从x-1开始遍历最长连续序列&#…...
设计模式-设计原则
文章目录 设计模式-设计原则单一职责原则开闭原则里氏替换原则依赖倒转原则接口隔离原则合成复用原则迪米特法则 设计模式-设计原则 单一职责原则 单一职责原则:一个对象应该只包含单一的职责,并且该职责被完整地封装在一个类中。 有时候类的功能并不…...
MongoDB基础运维
mongodb的基础概念介绍 database #数据库 collection #集合,类似于mysql中的表 filed #类似于mysql中字段 document #每行的记录 连接客户端 mongo ip:port 例如mongo 127.0.0.1:27017 mongo客户端的命…...
侧击雷如何检测预防
侧击雷是一种雷击的形式,指的是雷电从建筑物的侧面打来的直接雷击。侧击雷对高层建筑物的防雷保护提出了更高的要求,因为一般的避雷带或避雷针不能完全保护住建筑物的侧面。侧击雷可能会对建筑物的结构、设备和人员造成严重的损害,甚至引发火…...
检索搜索信息能力
(一)搜索工具的选择 谷歌 > 微信搜一搜 > 抖音等短视频 > 百度 (二)搜索方式 一,搜索内容的分类 信息类学习类工具类 二,谷歌浏览器的搜索技巧 1、“搜索内容” 限定完整出现的词 如下图搜…...
设计大咖亲授:Figma中文环境设置全攻略!
作为UI设计师,你一定很熟悉Figma,Figma是一款专注于UI/UX设计的在线协作工具,使用非常高效方便,不需要下载和安装。它只需要通过浏览器编辑,在国外很受欢迎。但是Figma对于国内的小伙伴来说,使用Figma有一定…...
华为Hcia-数通学习(更改策略)
方法:书籍视频题目训练 书籍:华为HCNA网络技术学习指南。 视频:网络工程师学习路线_哔哩哔哩_bilibili 有过一点基础。考软考网络工程师的时候做了大量笔记,回去复习了一遍。现在准备找工作了,开始学习华为认证的网…...
数据校验:Spring Validation
Spring Validation概述 在开发中,我们经常遇到参数校验的需求,比如用户注册的时候,要校验用户名不能为空、用户名长度不超过20个字符、手机号是合法的手机号格式等等。如果使用普通方式,我们会把校验的代码和真正的业务处理逻辑耦…...
CSS怎么选择除了第一个子元素外的其余同级子元素
使用 CSS 的:not()伪类选择器和:nth-child()伪类选择器 要通过CSS的代码选择某一个元素的除了第一个子元素外的其余的跟第一个子元素同级的子元素,可以结合使用CSS的:not()伪类选择器和:nth-child()伪类选择器进行选择。大致的语法如下: .parent > …...
Mac下eclipse配置JDK
目录 一、配置JDK,需要电脑下载Java并且配置环境 (1)、左上角找到“Eclipse”-->“Preferences...” (2)、找到“Java”-->“Installde JREs”-->界面显示电脑所安安装的Java;若没有需要点击“Add”进行配置 ①、选择“Standard VM”--&g…...
基于springboot实现体育场馆运营平台项目【项目源码】
基于springboot实现体育场馆运营管理系统演示 系统开发平台 在该数码论坛系统中,Eclipse能给用户提供更多的方便,其特点一是方便学习,方便快捷;二是有非常大的信息储存量,主要功能是用在对数据库中查询和编程。其功能…...
优雅的Java编程:将接口对象作为方法参数
theme: smartblue 目录 概述 在Java编程中,方法的参数传递方式通常是通过基本类型、对象引用或者集合等方式。然而,一种更加优雅且灵活的设计模式是将接口对象作为方法的参数。这种方式为我们带来了许多好处,包括降低耦合性、实现多态性和可…...
一文简单聊聊protobuf
目录 基本介绍 原理 同类对比 为什么要使用protobuf? 基本介绍 protobuf的全称是Protocol Buffer,是Google提供的一种数据序列化协议。Protocol Buffers 是一种轻便高效的结构化数据存储格式,可以用于结构化数据序列化,很适合做数据存储…...
Unity Meta Quest 一体机开发(五):手势抓取概述
文章目录 📕教程说明📕 Oculus Integration 中的三种手势抓取方式⭐Hand Grab⭐Touch Hand Grab⭐Distance Hand Grab 此教程相关的详细教案,文档,思维导图和工程文件会放入 Seed XR 社区。这是一个高质量知识星球 XR 社区&#…...
传输层中的TCP和UPD协议
一)应用层协议简介:根据需求明确要传输的信息,明确要传输的数据格式; 应用层协议:这个协议,实际上是和程序员打交道最多的协议了 1)其它四层都是操作系统,驱动,硬件实现好了的,咱们是不需要管 2)应用层:当我…...
插入排序算法(C++版)
1、什么是插入排序 插入排序(Insertion Sort)是一种简单直观的排序算法,它的基本思想是将一个待排序的数组分为已排序和未排序两个部分,然后逐步将未排序的元素插入到已排序的部分,直到整个数组有序。 2、插入排序的…...
(二)TensorRT-LLM | 模型导出(v0.20.0rc3)
0. 概述 上一节 对安装和使用有个基本介绍。根据这个 issue 的描述,后续 TensorRT-LLM 团队可能更专注于更新和维护 pytorch backend。但 tensorrt backend 作为先前一直开发的工作,其中包含了大量可以学习的地方。本文主要看看它导出模型的部分&#x…...
React19源码系列之 事件插件系统
事件类别 事件类型 定义 文档 Event Event 接口表示在 EventTarget 上出现的事件。 Event - Web API | MDN UIEvent UIEvent 接口表示简单的用户界面事件。 UIEvent - Web API | MDN KeyboardEvent KeyboardEvent 对象描述了用户与键盘的交互。 KeyboardEvent - Web…...
在鸿蒙HarmonyOS 5中使用DevEco Studio实现录音机应用
1. 项目配置与权限设置 1.1 配置module.json5 {"module": {"requestPermissions": [{"name": "ohos.permission.MICROPHONE","reason": "录音需要麦克风权限"},{"name": "ohos.permission.WRITE…...
docker 部署发现spring.profiles.active 问题
报错: org.springframework.boot.context.config.InvalidConfigDataPropertyException: Property spring.profiles.active imported from location class path resource [application-test.yml] is invalid in a profile specific resource [origin: class path re…...
SQL慢可能是触发了ring buffer
简介 最近在进行 postgresql 性能排查的时候,发现 PG 在某一个时间并行执行的 SQL 变得特别慢。最后通过监控监观察到并行发起得时间 buffers_alloc 就急速上升,且低水位伴随在整个慢 SQL,一直是 buferIO 的等待事件,此时也没有其他会话的争抢。SQL 虽然不是高效 SQL ,但…...
群晖NAS如何在虚拟机创建飞牛NAS
套件中心下载安装Virtual Machine Manager 创建虚拟机 配置虚拟机 飞牛官网下载 https://iso.liveupdate.fnnas.com/x86_64/trim/fnos-0.9.2-863.iso 群晖NAS如何在虚拟机创建飞牛NAS - 个人信息分享...
Linux部署私有文件管理系统MinIO
最近需要用到一个文件管理服务,但是又不想花钱,所以就想着自己搭建一个,刚好我们用的一个开源框架已经集成了MinIO,所以就选了这个 我这边对文件服务性能要求不是太高,单机版就可以 安装非常简单,几个命令就…...
Modbus RTU与Modbus TCP详解指南
目录 1. Modbus协议基础 1.1 什么是Modbus? 1.2 Modbus协议历史 1.3 Modbus协议族 1.4 Modbus通信模型 🎭 主从架构 🔄 请求响应模式 2. Modbus RTU详解 2.1 RTU是什么? 2.2 RTU物理层 🔌 连接方式 ⚡ 通信参数 2.3 RTU数据帧格式 📦 帧结构详解 🔍…...
区块链技术概述
区块链技术是一种去中心化、分布式账本技术,通过密码学、共识机制和智能合约等核心组件,实现数据不可篡改、透明可追溯的系统。 一、核心技术 1. 去中心化 特点:数据存储在网络中的多个节点(计算机),而非…...
SQL注入篇-sqlmap的配置和使用
在之前的皮卡丘靶场第五期SQL注入的内容中我们谈到了sqlmap,但是由于很多朋友看不了解命令行格式,所以是纯手动获取数据库信息的 接下来我们就用sqlmap来进行皮卡丘靶场的sql注入学习,链接:https://wwhc.lanzoue.com/ifJY32ybh6vc…...
