AI 助理
文档备案控制台
开发者社区 大数据与机器学习 大数据计算 MaxCompute 文章 正文

UDF精简使用大全

2021-06-07 6730
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 在MaxCompute开发过程中，开发同学遇到的的一些复杂逻辑该如何处理，如何在MaxCompute开发不同场景下的UDF函数？带着这个问题，本文针对UDF的各方面内容做出介绍，其中涉及UDF对应不同语言的类型映射关系、以及对应UDF在重载、访问网络、引用表与资源、以及第三方包的使用为大家做出展示。

一、UDF的分类

UDF类型

描述

UDF（User Defined Scalar Function）

用户自定义标量值函数。其输入与输出是一对一的关系，即读入一行数据，输出一个值。

UDTF（User Defined Table Valued Function）

自定义表值函数。用于解决调用一次函数输出多行数据的需求。UDTF是唯一能够返回多个字段的自定义函数。UDTF不等于UDT（User Defined Type）。

UDAF（User Defined Aggregation Function）

自定义聚合函数。其输入与输出是多对一的关系，即将多条输入记录聚合成一个输出值。UDAF可以与SQL中的GROUP BY语句联用。具体语法请参见

聚合函数

。

二、UDF参数解析

MaxCompute数据类型与Java数据类型的对应关系如下。

注意点：

  • 此处ARRAY类型对应的Java类型是List，而不是数组。
  • VARCHAR,BINART,STRUCT一些数据类型是ODPS独有的
  • Java中对应的数据类型以及返回值数据类型是对象，数据类型首字母需大写。

MaxCompute Type

Java Type

TINYINT

java.lang.Byte

SMALLINT

java.lang.Short

INT

java.lang.Integer

BIGINT

java.lang.Long

FLOAT

java.lang.Float

DOUBLE

java.lang.Double

DECIMAL

java.math.BigDecimal

BOOLEAN

java.lang.Boolean

STRING

java.lang.String

VARCHAR

com.aliyun.odps.data.Varchar

BINARY

com.aliyun.odps.data.Binary

DATETIME

java.util.Date

TIMESTAMP

java.sql.Timestamp

ARRAY

java.util.List

MAP

java.util.Map

STRUCT

com.aliyun.odps.data.Struct

MaxCompute 2.0版本支持定义Java UDF时，使用Writable类型作为参数和返回值。MaxCompute数据类型和Java Writable类型的映射关系如下。

MaxCompute Type

Java Writable Type

TINYINT

ByteWritable

SMALLINT

ShortWritable

INT

IntWritable

BIGINT

LongWritable

FLOAT

FloatWritable

DOUBLE

DoubleWritable

DECIMAL

BigDecimalWritable

BOOLEAN

BooleanWritable

STRING

Text

VARCHAR

VarcharWritable

BINARY

BytesWritable

DATETIME

DatetimeWritable

TIMESTAMP

TimestampWritable

INTERVAL_YEAR_MONTH

IntervalYearMonthWritable

INTERVAL_DAY_TIME

IntervalDayTimeWritable

ARRAY

N/A

MAP

N/A

STRUCT

N/A

MaxCompute SQL Type

Python 2 Type

BIGINT

INT

STRING

STR

DOUBLE

FLOAT

BOOLEAN

BOOL

DATETIME

INT

FLOAT

FLOAT

CHAR

STR

VARCHAR

STR

BINARY

BYTEARRAY

DATE

INT

DECIMAL

DECIMAL.DECIMAL

ARRAY

LIST

MAP

DICT

STRUCT

COLLECTIONS.NAMEDTUPLE

MaxCompute SQL Type

Python 3 Type

BIGINT

INT

STRING

UNICODE

DOUBLE

FLOAT

BOOLEAN

BOOL

DATETIME

DATETIME.DATETIME

FLOAT

FLOAT

CHAR

UNICODE

VARCHAR

UNICODE

BINARY

BYTES

DATE

DATETIME.DATE

DECIMAL

DECIMAL.DECIMAL

ARRAY

LIST

MAP

DICT

STRUCT

COLLECTIONS.NAMEDTUPLE

三、UDF的使用方式

UDF、UDTF、UDAT可进行参考文档

https://help.aliyun.com/document_detail/27867.html?spm=a2c4g.11186623.6.762.463d7468xnFPHb

JAVA UDF

UDF的高级使用：

3.1UDF中的变长参数

java语言：

packagecom.mrtest.cn;
importcom.aliyun.odps.udf.UDF;
importcom.aliyun.odps.udf.annotation.Resolve;
importjava.util.ArrayList;
importjava.util.List;
@Resolve({"*->array"})
publicclassTestUDFextendsUDF {
publicListevaluate(String ... s) {
Listlist=newArrayList();
for (Stringname : s) {
list.add(name);
        }
returnlist;
    }
}

Python语言：

fromodps.udfimportannotate@annotate("*->bigint")
classParamFunc(object):
defevaluate(self, *nums):
sum=0fornuminnums:
sum=num+sumreturnsum

3.2UDF的重载

注意事项：对于List与List是不能解析对应的方法的，这种属于类型擦除

packagecom.aliyun.odps.examples.udf;
importcom.aliyun.odps.udf.UDF;
publicclassUDFExampleextendsUDF {
publicStringevaluate(Stringa) {
return"s2s:"+a;
  }
publicStringevaluate(Stringa, Stringb) {
return"ss2s:"+a+","+b;
  }
publicStringevaluate(Stringa, Stringb, Stringc) {
return"sss2s:"+a+","+b+","+c;
  }
}

3.3UDF访问对应文件和表

java语言：

packagecom.aliyun.odps.examples.udf;
importcom.aliyun.odps.udf.ExecutionContext;
importcom.aliyun.odps.udf.UDF;
importcom.aliyun.odps.udf.UDFException;
importjava.io.BufferedReader;
importjava.io.IOException;
importjava.io.InputStream;
importjava.io.InputStreamReader;
importjava.util.Iterator;
publicclassUDFResourceextendsUDF {
ExecutionContextctx;
longfileResourceLineCount;
longtableResource1RecordCount;
longtableResource2RecordCount;
@Overridepublicvoidsetup(ExecutionContextctx) throwsUDFException {
this.ctx=ctx;
try {
InputStreamin=ctx.readResourceFileAsStream("file_resource.txt");
BufferedReaderbr=newBufferedReader(newInputStreamReader(in));
Stringline;
fileResourceLineCount=0;
while ((line=br.readLine()) !=null) {
fileResourceLineCount++;
      }
br.close();
Iteratoriterator=ctx.readResourceTable("table_resource1").iterator();
tableResource1RecordCount=0;
while (iterator.hasNext()) {
tableResource1RecordCount++;
iterator.next();
      }
iterator=ctx.readResourceTable("table_resource2").iterator();
tableResource2RecordCount=0;
while (iterator.hasNext()) {
tableResource2RecordCount++;
iterator.next();
      }
    } catch (IOExceptione) {
thrownewUDFException(e);
    }
  }
/*** project: example_project table: wc_in2 partitions: p2=1,p1=2 columns: colc,colb*/publicStringevaluate(Stringa, Stringb) {
return"ss2s:"+a+","+b+"|fileResourceLineCount="+fileResourceLineCount+"|tableResource1RecordCount="+tableResource1RecordCount+"|tableResource2RecordCount="+tableResource2RecordCount;
  }
}



python语言：

#coding: utf-8fromodps.udfimportannotatefromodps.distcacheimportget_cache_file@annotate('double -> double')
classCompute(object):
def__init__(self):
importjson#获取对应文本文件cache_file=get_cache_file('file.txt')
dataMat= []
forlineincache_file :
curLine=line.strip().split(',')
#处理逻辑cache_file.close()
#获取对应的表文件records=list(get_cache_table('table_resource1'))
forrecordinrecords:
self.my_dict[record[0]] = [record[1]]
#处理逻辑defevaluate(self, input):
#处理逻辑

     

3.4UDF访问外部网络(VPC、外部网络、专有网络）

https://help.aliyun.com/document_detail/187866.html

3.5UDF使用第三方包

https://help.aliyun.com/document_detail/189752.html

#coding: utf-8# explode.pyfromodps.udfimportannotatefromodps.distcacheimportget_cache_archiveimportdatetimedefinclude_package_path(res_name):
importos, sysarchive_files=get_cache_archive(res_name)
dir_names=sorted ([os.path.dirname(os.path.normpath(f.name)) forfinarchive_filesif'.dist_info'notinf.name], key=lambdav: len(v))
sys.path.append(os.path.dirname(dir_names[0]))
@annotate("string->boolean")
classis_workday_udf(object):
def__init__(self):
include_package_path('chinese-calendar-master.zip')
defevaluate(self, date_str):
# try:importchinese_calendardate_strs=date_str.split("-")
year_num=int(date_strs[0])
month_num=int(date_strs[1])
day_num=int(date_strs[2])
date_num=datetime.date(year=year_num, month=month_num, day=day_num)
result=chinese_calendar.is_workday(date_num)
returnresult# except:#     return True

函数的注册

执行的select的的操作

set odsp.pypy.enabled=false;set odps.isolation.session.enable=true;select my_json('{"info":"11","desc":"a|b","filename":"4b-2a-3c-4d-5b"}')as a;



3.6使用嵌入式开发UDF

java语言：

CREATE TEMPORARY FUNCTION foo AS'com.mypackage.Reverse' USING
#CODE ('lang'='JAVA')package com.mypackage;import com.aliyun.odps.udf.UDF;public class Reverse extends UDF {  public String evaluate(String input){    if (input ==null) return null;    StringBuilder ret = new StringBuilder();    for (int i = input.toCharArray().length-1; i >=0; i--) {      ret.append(input.toCharArray()[i]);}    return ret.toString();}}#END CODE;


SELECT foo('abdc');

  • 嵌入式代码块可以置于USING后或脚本末尾，置于USING后的代码块作用域仅为CREATE TEMPORARY FUNCTION语句。
  • CREATE TEMPORARY FUNCTION创建的函数为临时函数，仅在本次执行生效，不会存入MaxCompute的Meta系统。


python语言：

CREATETEMPORARYFUNCTIONfooAS'embedded.UDFTest'USING#CODE ('lang'='PYTHON', 'filename'='embedded')fromodps.udfimportannotate@annotate("bigint->bigint")
classUDFTest(object):
defevaluate(self, a):
returna*a#END CODE;


SELECT foo(4);

  • Python代码的缩进需要符合Python语言规范。
  • 由于注册Python UDF时AS后的类名需要包含Python源码的文件名，您可以通过’filename’=’embedded’指定一个虚拟文件名。

3.7使用SQL语言定义函数

create sql function my_sum(@a BIGINT, @b BIGINT, @c BIGINT) returns @my_sum BIGINTasbegin    @temp := @a + @b;    @my_sum := @temp + @c;end;
create sql function my_func(@s STRING)AS if(@s rlike '"git_(m|a)"',1,0);


欢迎加入“MaxCompute开发者社区2群”,点击链接申请加入或扫描二维码

https://h5.dingtalk.com/invite-page/index.html?bizSource=____source____&corpId=dingb682fb31ec15e09f35c2f4657eb6378f&inviterUid=E3F28CD2308408A8&encodeDeptId=0054DC2B53AFE745

文章标签：
云原生大数据计算服务 MaxCompute
Java
Python
分布式计算
SQL
网络安全
数据格式
开发者
MaxCompute
JSON
相关实践学习
基于MaxCompute的热门话题分析
Apsara Clouder大数据专项技能认证配套课程：基于MaxCompute的热门话题分析
耿江涛
目录
相关文章
探索云世界
|
7月前
|
SQL JSON 分布式计算
【跨国数仓迁移最佳实践6】MaxCompute SQL语法及函数功能增强，10万条SQL转写顺利迁移
本系列文章将围绕东南亚头部科技集团的真实迁移历程展开，逐步拆解 BigQuery 迁移至 MaxCompute 过程中的关键挑战与技术创新。本篇为第六篇，MaxCompute SQL语法及函数功能增强。 注：客户背景为东南亚头部科技集团，文中用 GoTerra 表示。
探索云世界
789 20
赵渝强老师
|
11月前
|
存储 分布式计算 大数据
【赵渝强老师】阿里云大数据存储计算服务：MaxCompute
阿里云MaxCompute是快速、全托管的TB/PB级数据仓库解决方案，提供海量数据存储与计算服务。支持多种计算模型，适用于大规模离线数据分析，具备高安全性、低成本、易用性强等特点，助力企业高效处理大数据。
赵渝强老师
505 0
阿里云开发者
|
9月前
|
SQL 存储 分布式计算
【万字长文，建议收藏】《高性能ODPS SQL章法》——用古人智慧驾驭大数据战场
本文旨在帮助非专业数据研发但是有高频ODPS使用需求的同学们（如数分、算法、产品等）能够快速上手ODPS查询优化，实现高性能查数看数，避免日常工作中因SQL任务卡壳、失败等情况造成的工作产出delay甚至集群资源稳定性问题。
阿里云开发者
1625 36
【万字长文，建议收藏】《高性能ODPS SQL章法》——用古人智慧驾驭大数据战场
SHIYUANSHI
|
11月前
|
SQL 分布式计算 DataWorks
我与阿里云ODPS的故事：从挑战到掌控
本文讲述了作者在使用阿里云ODPS过程中的亲身经历，从最初面对大数据处理的困境，到通过ODPS及其核心组件MaxCompute、DataWorks实现数据处理与开发效率的大幅提升。不仅展现了技术带来的变革，也体现了个人从挑战到掌控的成长历程。
SHIYUANSHI
295 10
赵渝强老师
|
9月前
|
存储 分布式计算 资源调度
【赵渝强老师】阿里云大数据MaxCompute的体系架构
阿里云MaxCompute是快速、全托管的EB级数据仓库解决方案，适用于离线计算场景。它由计算与存储层、逻辑层、接入层和客户端四部分组成，支持多种计算任务的统一调度与管理。
赵渝强老师
754 1
taro_秋刀鱼
|
SQL 分布式计算 DataWorks
使用DataWorks PyODPS节点调用XGBoost算法
本文介绍如何在DataWorks中通过PyODPS3节点调用XGBoost算法完成模型训练与测试，并实现周期离线调度。主要内容包括：1) 使用ODPS SQL构建数据集；2) 创建PyODPS3节点进行数据处理与模型训练；3) 构建支持XGBoost的自定义镜像；4) 测试运行并选择对应镜像。适用于需要集成机器学习算法到大数据工作流的用户。
taro_秋刀鱼
520 24
阿里云云原生
|
人工智能 前端开发 开发工具
对话阿里云通义灵码技术负责人陈鑫：AI编程的现状与未来
在AI快速发展的2025年，通义灵码作为国内领先的AI编程助手，正通过其独特的智能体架构和强大模型能力重新定义开发方式。本文邀请技术负责人陈鑫（神秀），探讨AI编程现状与未来。通义灵码基于Qwen3模型打造，具备记忆系统革新、MCP工具生态和多模态交互等优势，推出三种工作模式以适应不同场景。尽管行业仍面临挑战，但国产模型正在崛起，企业可采用“三步走”策略引入AI工具。未来，AI将从辅助走向主导，深化代码理解并重构开发工具，助力更高效、创造性的编程方式。
阿里云云原生
1054 7
扬流
|
人工智能 分布式计算 大数据
云栖实录 | MaxCompute 迈向下一代的智能云数仓
2024年云栖大会上，阿里云核心自研云原生智能数据仓库产品MaxCompute，在经过一年的深度打磨后，推出了其迈向下一代智能云数据仓的系列主题分享。此次产品发布，充分展示MaxCompute产品领先行业的云数据产品发展理念与核心优势。
扬流
1926 2
花开富贵111
|
分布式计算 MaxCompute 计算机视觉
ODPS问题之odps.sql.mapper.split.size属性有什么作用，以及如何根据场景调整它
ODPS问题之odps.sql.mapper.split.size属性有什么作用，以及如何根据场景调整它
花开富贵111
1532 1
嘟嘟嘟嘟嘟嘟
|
分布式计算 运维 DataWorks
DataWorks产品使用合集之如何实现任务的批量导入和导出
DataWorks作为一站式的数据开发与治理平台，提供了从数据采集、清洗、开发、调度、服务化、质量监控到安全管理的全套解决方案，帮助企业构建高效、规范、安全的大数据处理体系。以下是对DataWorks产品使用合集的概述，涵盖数据处理的各个环节。
嘟嘟嘟嘟嘟嘟
570 0
大数据与机器学习

大数据计算 MaxCompute

热门文章

最新文章

  • 1
    odps是什么?
  • 2
    MaxCompute执行作业慢的原因排查
  • 3
    阿里云MaxCompute（大数据）公开数据集---带你玩转人工智能
  • 4
    吴刚专访--大数据和 MaxCompute 技术和故事
  • 5
    [大数据新手上路]“零基础”系列课程--如何将ECS上的Hadoop数据迁移到阿里云数加·MaxCompute
  • 6
    干货：解码OneData，阿里的数仓之路。
  • 7
    使用 MaxCompute Studio 开发大数据应用
  • 8
    ［ETL实践指南］基于Kettle的MaxCompute插件实现数据上云
  • 9
    混合云模式下 MaxCompute + Hadoop 混搭大数据架构实践
  • 10
    【内含分享PPT/视频/文章】阿里云MVP学院MaxCompute技术闭门会线上首播 | 2019大数据技术公开课第二季
  • 1
    #Nginx教程 Nginx作为目前最流行的高性能Web服务器和反向代理服务器，凭借其高并发、低内存消耗的特点，被广泛应用于各类生产环境。本文将从零开始，带你快速掌握Nginx的核心配置与实战技巧。
    328
  • 2
    金融行情系统中，API 接入常见的 5 个工程问题
    286
  • 3
    寻找 AI 全能王——阿里云 Data+AI 工程师全球大奖赛正式开启
    1305
  • 4
    网站代码 网站源代码 网页源代码 网页代码网站
    1855
  • 5
    【跨国数仓迁移最佳实践8】MaxCompute Streaming Insert：大数据数据流写业务迁移的实践与突破
    1769
  • 6
    【跨国数仓迁移最佳实践7】基于 MaxCompute 多租的大数据平台架构
    468
  • 7
    ODPS 十五周年实录 | Data + AI，MaxCompute 下一个15年的新增长引擎
    588
  • 8
    【跨国数仓迁移最佳实践6】MaxCompute SQL语法及函数功能增强，10万条SQL转写顺利迁移
    789
  • 9
    诊断设备企业必看！垂直医疗行业的CRM软件有哪些？
    481
  • 10
    MaxCompute 聚簇优化推荐原理
    507

    • 相关电子书

    更多
  • 低代码开发师（初级）实战教程
  • 冬季实战营第三期：MySQL数据库进阶实战
  • 阿里巴巴DevOps 最佳实践手册
    • 下一篇
    阿里云正式发布 Agentic 代码安全：AI驱动的双Agent协同引擎