当前位置: 首页 > news >正文

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)

使用场景&#xff1a; 标量函数即 UDF&#xff0c;⽤于进⼀条数据出⼀条数据的场景。 开发流程&#xff1a; 实现 org.apache.flink.table.functions.ScalarFunction 接⼝实现⼀个或者多个⾃定义的 eval 函数&#xff0c;名称必须叫做 eval&#xff0c;eval ⽅法签名必须是 p…...

【第2章 Node.js基础】2.4 Node.js 全局对象(二) process 对象

process对象是一个全局对象&#xff0c;提供当前Node.js 进程信息并对其进行控制。通常用于编写本地命令行程序。 1.进程事件 process对象是EventEmitter类的实例&#xff0c;因此可以使用事件的方式来处理和监听process对象的各种事件。以下是一些常用的process对象事件&…...

觉非科技发布【轻地图高速NOA智驾方案】|地平线,觉非科技,MobileDrive超捷生态协作实现技术落地

11月10日&#xff0c;2023世界新汽车技术合作生态展期间&#xff0c;觉非科技正式发布了基于地平线征程系列以及MobileDrive超捷规控系统的【轻地图高速NOA智驾解决方案】。该方案首次实现了从芯片能力输出、到数据闭环建设、再到规控部署的产业生态链协作与打通&#xff0c;为…...

竞赛 车道线检测(自动驾驶 机器视觉)

0 前言 无人驾驶技术是机器学习为主的一门前沿领域&#xff0c;在无人驾驶领域中机器学习的各种算法随处可见&#xff0c;今天学长给大家介绍无人驾驶技术中的车道线检测。 1 车道线检测 在无人驾驶领域每一个任务都是相当复杂&#xff0c;看上去无从下手。那么面对这样极其…...

128. 最长连续序列

这道题最简单的想法就是排序计数&#xff0c;但是复杂度为O(nlogn)&#xff0c;不符合题意 于是采用哈希表的方法 将所有数字存放在哈希表中&#xff0c;然后开始逐个寻找。 比如当前遍历到x&#xff0c;如果x-1也存在哈希表中&#xff0c;那就从x-1开始遍历最长连续序列&#…...

设计模式-设计原则

文章目录 设计模式-设计原则单一职责原则开闭原则里氏替换原则依赖倒转原则接口隔离原则合成复用原则迪米特法则 设计模式-设计原则 单一职责原则 单一职责原则&#xff1a;一个对象应该只包含单一的职责&#xff0c;并且该职责被完整地封装在一个类中。 有时候类的功能并不…...

MongoDB基础运维

mongodb的基础概念介绍 database #数据库 collection #集合&#xff0c;类似于mysql中的表 filed #类似于mysql中字段 document #每行的记录 连接客户端 mongo ip:port 例如mongo 127.0.0.1:27017 mongo客户端的命…...

侧击雷如何检测预防

侧击雷是一种雷击的形式&#xff0c;指的是雷电从建筑物的侧面打来的直接雷击。侧击雷对高层建筑物的防雷保护提出了更高的要求&#xff0c;因为一般的避雷带或避雷针不能完全保护住建筑物的侧面。侧击雷可能会对建筑物的结构、设备和人员造成严重的损害&#xff0c;甚至引发火…...

检索搜索信息能力

&#xff08;一&#xff09;搜索工具的选择 谷歌 > 微信搜一搜 > 抖音等短视频 > 百度 &#xff08;二&#xff09;搜索方式 一&#xff0c;搜索内容的分类 信息类学习类工具类 二&#xff0c;谷歌浏览器的搜索技巧 1、“搜索内容” 限定完整出现的词 如下图搜…...

设计大咖亲授:Figma中文环境设置全攻略!

作为UI设计师&#xff0c;你一定很熟悉Figma&#xff0c;Figma是一款专注于UI/UX设计的在线协作工具&#xff0c;使用非常高效方便&#xff0c;不需要下载和安装。它只需要通过浏览器编辑&#xff0c;在国外很受欢迎。但是Figma对于国内的小伙伴来说&#xff0c;使用Figma有一定…...

华为Hcia-数通学习(更改策略)

方法&#xff1a;书籍视频题目训练 书籍&#xff1a;华为HCNA网络技术学习指南。 视频&#xff1a;网络工程师学习路线_哔哩哔哩_bilibili 有过一点基础。考软考网络工程师的时候做了大量笔记&#xff0c;回去复习了一遍。现在准备找工作了&#xff0c;开始学习华为认证的网…...

数据校验:Spring Validation

Spring Validation概述 在开发中&#xff0c;我们经常遇到参数校验的需求&#xff0c;比如用户注册的时候&#xff0c;要校验用户名不能为空、用户名长度不超过20个字符、手机号是合法的手机号格式等等。如果使用普通方式&#xff0c;我们会把校验的代码和真正的业务处理逻辑耦…...

CSS怎么选择除了第一个子元素外的其余同级子元素

使用 CSS 的:not()伪类选择器和:nth-child()伪类选择器 要通过CSS的代码选择某一个元素的除了第一个子元素外的其余的跟第一个子元素同级的子元素&#xff0c;可以结合使用CSS的:not()伪类选择器和:nth-child()伪类选择器进行选择。大致的语法如下&#xff1a; .parent > …...

Mac下eclipse配置JDK

目录 一、配置JDK&#xff0c;需要电脑下载Java并且配置环境 (1)、左上角找到“Eclipse”-->“Preferences...”​ (2)、找到“Java”-->“Installde JREs”-->界面显示电脑所安安装的Java&#xff1b;若没有需要点击“Add”进行配置​ ①、选择“Standard VM”--&g…...

基于springboot实现体育场馆运营平台项目【项目源码】

基于springboot实现体育场馆运营管理系统演示 系统开发平台 在该数码论坛系统中&#xff0c;Eclipse能给用户提供更多的方便&#xff0c;其特点一是方便学习&#xff0c;方便快捷&#xff1b;二是有非常大的信息储存量&#xff0c;主要功能是用在对数据库中查询和编程。其功能…...

优雅的Java编程:将接口对象作为方法参数

theme: smartblue 目录 概述 在Java编程中&#xff0c;方法的参数传递方式通常是通过基本类型、对象引用或者集合等方式。然而&#xff0c;一种更加优雅且灵活的设计模式是将接口对象作为方法的参数。这种方式为我们带来了许多好处&#xff0c;包括降低耦合性、实现多态性和可…...

一文简单聊聊protobuf

目录 基本介绍 原理 同类对比 为什么要使用protobuf? 基本介绍 protobuf的全称是Protocol Buffer&#xff0c;是Google提供的一种数据序列化协议。Protocol Buffers 是一种轻便高效的结构化数据存储格式&#xff0c;可以用于结构化数据序列化&#xff0c;很适合做数据存储…...

Unity Meta Quest 一体机开发(五):手势抓取概述

文章目录 &#x1f4d5;教程说明&#x1f4d5; Oculus Integration 中的三种手势抓取方式⭐Hand Grab⭐Touch Hand Grab⭐Distance Hand Grab 此教程相关的详细教案&#xff0c;文档&#xff0c;思维导图和工程文件会放入 Seed XR 社区。这是一个高质量知识星球 XR 社区&#…...

传输层中的TCP和UPD协议

一)应用层协议简介:根据需求明确要传输的信息&#xff0c;明确要传输的数据格式&#xff1b; 应用层协议:这个协议&#xff0c;实际上是和程序员打交道最多的协议了 1)其它四层都是操作系统&#xff0c;驱动&#xff0c;硬件实现好了的&#xff0c;咱们是不需要管 2)应用层:当我…...

插入排序算法(C++版)

1、什么是插入排序 插入排序&#xff08;Insertion Sort&#xff09;是一种简单直观的排序算法&#xff0c;它的基本思想是将一个待排序的数组分为已排序和未排序两个部分&#xff0c;然后逐步将未排序的元素插入到已排序的部分&#xff0c;直到整个数组有序。 2、插入排序的…...

深入剖析AI大模型:大模型时代的 Prompt 工程全解析

今天聊的内容&#xff0c;我认为是AI开发里面非常重要的内容。它在AI开发里无处不在&#xff0c;当你对 AI 助手说 "用李白的风格写一首关于人工智能的诗"&#xff0c;或者让翻译模型 "将这段合同翻译成商务日语" 时&#xff0c;输入的这句话就是 Prompt。…...

c++ 面试题(1)-----深度优先搜索(DFS)实现

操作系统&#xff1a;ubuntu22.04 IDE:Visual Studio Code 编程语言&#xff1a;C11 题目描述 地上有一个 m 行 n 列的方格&#xff0c;从坐标 [0,0] 起始。一个机器人可以从某一格移动到上下左右四个格子&#xff0c;但不能进入行坐标和列坐标的数位之和大于 k 的格子。 例…...

零基础设计模式——行为型模式 - 责任链模式

第四部分&#xff1a;行为型模式 - 责任链模式 (Chain of Responsibility Pattern) 欢迎来到行为型模式的学习&#xff01;行为型模式关注对象之间的职责分配、算法封装和对象间的交互。我们将学习的第一个行为型模式是责任链模式。 核心思想&#xff1a;使多个对象都有机会处…...

保姆级教程:在无网络无显卡的Windows电脑的vscode本地部署deepseek

文章目录 1 前言2 部署流程2.1 准备工作2.2 Ollama2.2.1 使用有网络的电脑下载Ollama2.2.2 安装Ollama&#xff08;有网络的电脑&#xff09;2.2.3 安装Ollama&#xff08;无网络的电脑&#xff09;2.2.4 安装验证2.2.5 修改大模型安装位置2.2.6 下载Deepseek模型 2.3 将deepse…...

Python Ovito统计金刚石结构数量

大家好,我是小马老师。 本文介绍python ovito方法统计金刚石结构的方法。 Ovito Identify diamond structure命令可以识别和统计金刚石结构,但是无法直接输出结构的变化情况。 本文使用python调用ovito包的方法,可以持续统计各步的金刚石结构,具体代码如下: from ovito…...

FFmpeg avformat_open_input函数分析

函数内部的总体流程如下&#xff1a; avformat_open_input 精简后的代码如下&#xff1a; int avformat_open_input(AVFormatContext **ps, const char *filename,ff_const59 AVInputFormat *fmt, AVDictionary **options) {AVFormatContext *s *ps;int i, ret 0;AVDictio…...

内窥镜检查中基于提示的息肉分割|文献速递-深度学习医疗AI最新文献

Title 题目 Prompt-based polyp segmentation during endoscopy 内窥镜检查中基于提示的息肉分割 01 文献速递介绍 以下是对这段英文内容的中文翻译&#xff1a; ### 胃肠道癌症的发病率呈上升趋势&#xff0c;且有年轻化倾向&#xff08;Bray等人&#xff0c;2018&#x…...

【版本控制】GitHub Desktop 入门教程与开源协作全流程解析

目录 0 引言1 GitHub Desktop 入门教程1.1 安装与基础配置1.2 核心功能使用指南仓库管理日常开发流程分支管理 2 GitHub 开源协作流程详解2.1 Fork & Pull Request 模型2.2 完整协作流程步骤步骤 1: Fork&#xff08;创建个人副本&#xff09;步骤 2: Clone&#xff08;克隆…...

python基础语法Ⅰ

python基础语法Ⅰ 常量和表达式变量是什么变量的语法1.定义变量使用变量 变量的类型1.整数2.浮点数(小数)3.字符串4.布尔5.其他 动态类型特征注释注释是什么注释的语法1.行注释2.文档字符串 注释的规范 常量和表达式 我们可以把python当作一个计算器&#xff0c;来进行一些算术…...

IP选择注意事项

IP选择注意事项 MTP、FTP、EFUSE、EMEMORY选择时&#xff0c;需要考虑以下参数&#xff0c;然后确定后选择IP。 容量工作电压范围温度范围擦除、烧写速度/耗时读取所有bit的时间待机功耗擦写、烧写功耗面积所需要的mask layer...