MaxCompute - ODPS重装上阵 第八弹 - 动态类型函数

本文涉及的产品
云原生大数据计算服务 MaxCompute,5000CU*H 100GB 3个月
云原生大数据计算服务MaxCompute,500CU*H 100GB 3个月
简介: MaxCompute自定义函数的参数和返回值不够灵活,是数据开发过程中时常被提及的问题。Hive 提供给了 GenericUDF 的方式,通过调用一段用户代码,让用户来根据参数类型决定返回值类型。MaxCompute 出于性能、安全性等考虑,没有支持这种方式。

MaxCompute(原ODPS)是阿里云自主研发的具有业界领先水平的分布式大数据处理平台, 尤其在集团内部得到广泛应用,支撑了多个BU的核心业务。 MaxCompute除了持续优化性能外,也致力于提升SQL语言的用户体验和表达能力,提高广大ODPS开发者的生产力。

MaxCompute基于ODPS2.0新一代的SQL引擎,显著提升了SQL语言编译过程的易用性与语言的表达能力。我们在此推出MaxCompute(ODPS2.0)重装上阵系列文章

MaxCompute自定义函数的参数和返回值不够灵活,是数据开发过程中时常被提及的问题。Hive 提供给了 GenericUDF 的方式,通过调用一段用户代码,让用户来根据参数类型决定返回值类型。MaxCompute 出于性能、安全性等考虑,没有支持这种方式。但是MaxCompute也提供了许多方式,让您能够灵活地自定义函数。

  • 场景1
    需要实现一个UDF,可以接受任意类型的输入,但是MaxCompute的UDF不支持泛型,要做一个接受任何类型的函数,就必须为每种类型都写一个evaluate函数。
  • 场景2
    MaxCompute的UDAF和UDTF使用@Resolve的注解来指定输入输出类型,无法重载。要做一个接受多种类型的自定义功能,就需要定义多个不同的函数。
  • 场景3
    MaxCompute支持了参数化视图,能够把一些公共的SQL提出来。参数化视图的表值参数要求输入表的列数和类型与视图定义时完全一致,如果想要写一个能够接受具有相似特征的不同的表的视图,还无法定义出来。

本文带大家一起看看MaxCompute对这些大家关心的问题都做了哪些改进。

参数化视图

问题

参数化视图是MaxCompute自己设计的一种视图。允许用户定义参数,从而能够大大视图代码的复用率。很多用户都利用这一功能,将一些公共SQL提取到视图中,形成公共SQL代码池。

参数化视图在声明过程中具有局限性:参数类型,长度都是固定的。尤其是参数化视图允许传入表值参数,表值参数要求形参与实参在列的个数和类型上都一致。这一点限制了许多使用场景,如下面的例子:

CREATE VIEW paramed_view (@a TABLE(key bigint)) AS SELECT @a.* FROM @a JOIN dim on a.key = dim.key;

这个例子封装了一段使用dim表来过滤输入表的逻辑,本来这个是个通用的逻辑,任何包含key这一列的表,都可以用来做输入表。但是由于定义视图时只能确定输入中包含key列,因此声明的参数类型只包含这一列。导致了视图的调用者传递的表参数必须只能有一列,而返回的数据集也只包含一列,这显然与这个视图的设计初衷不合。

改进

最新的MaxCompute版本对参数化视图做了一些改进,可以大大提升参数化视图定义的灵活性。

首先,参数化视图的参数可以使用ANY关键字,表示任意类型。如

CREATE VIEW paramed_view (@a ANY) AS SELECT * FROM src WHERE case when @a is null then key1 else key2 end = key3;

这里定义的视图,第一个参数可以接受任意类型。注意ANY类型不能参与如 '+', 'AND' 之类的需要明确类型才能做的运算。ANY类型更多是在TABLE参数中做passthrough列,如

CREATE VIEW paramed_view (@a TABLE(name STRING, id ANY, age BIGINT)) AS SELECT * FROM @a WHRER name = 'foo' and age < 25;

-- 调用示例
SELECT * FROM param_view((SELECT name, id, age from students));

上面的视图接受一个表值参数,但是并不关心这个表的第二列,那么这个列可以直接定义为ANY类型。参数化视图在调用时,每次都会根据输入参数的实际类型重新推算返回值类型。比如上面的视图,当输入的表是 TABLE(c1 STRING, c2 DOUBLE, c3 BIGINT),那么输出的数据集的第二列也会自动变成DOUBLE类型,让视图的调用者可以使用任何可用于DOUBLE类型的操作来操作这一列。

需要注意的一点是,我们用CREATE VIEW创建了视图后,可以用DESC来获取视图的描述,这个描述中会包含视图的返回类型信息。但是由于视图的返回类型是在调用的时候重新推算的,重新推算出来的类型可能与创建视图时推导出来的不一致。一个例子就是上面的ANY类型。

在ANY之外,参数化视图中的表值参数还支持了*,表示任意多个列。这个 * 可以带类型,也可以使用ANY类型。如

CREATE VIEW paramed_view (@a TABLE(key STRING, * ANY), @b TABLE(key STRING, * STRING)) AS SELECT a.* FROM @a JOIN @b ON a.key = b.key; 

-- 调用示例
SELECT name, address FROM param_view((SELECT school, name, age, address FROM student), school) WHERE age < 20;

上面这个视图接受两个表值参数,第一个表值参数第一列是string类型,后面可以是任意多个任意类型的列,而第二个表值参数的第一列是string,后面可以是任意多个STRING类型的列。这其中有几点需要注意:

  • 变长部分必须要写在表值参数定义的最后面,即在 * 的后面不允许再有其他列。这也间接导致了一个表值参数中最多只有一个变长列列表。
  • 由于变长部分必须在最后,有的时候输入表的列不一定是按照这种顺序排列的,这时候需要对输入表的列做一定重排,可以以subquery作为参数(参考上面的例子),注意subquery外面要加一层括号。
  • 由于表值参数中变长部分没有名字,因此在视图定义过程中没办法获得对这部分数据的引用,也就没有办法对这些数据做运算。这个限制是特意设置的,如果需要对变长部分的数据做运算,需要把要运算的列声明在定长部分,而编译器会对调用时传入的参数进行检查。
  • 虽然不能对变长部分做运算,但是 SELECT * 这种通配符的使用依旧可以将变长部分的列传递出去,如上面的例子在paramed_view中将 @a 的所有列返回,虽然创建视图的时候,a中只有key这一列,但是调用视图的时候,编译器推算出@a中还包含了name, age, address,因此视图返回的数据集中也包含这三列,而视图的调用者也可以对着三列进行操作(如 WHERE age < 20)。
  • 表值参数的列与视图声明时指定的定长列部分不一定完全一致。如果名字不一样,编译器会自动做重命名,如果类型不一样,编译器会做隐式转换(不能隐式转换则会报错)。

上面提到的第4点非常有用,一方面保证了调用视图是输入参数的灵活性,另一方面又不降低数据的信息量。好好利用能够很大程度上增加公共代码的复用率。

下面是一个调用示例。该例子使用的视图是:

CREATE VIEW paramed_view (@a TABLE(key STRING, * ANY), @b TABLE(key STRING, * STRING)) AS SELECT a.* FROM @a JOIN @b ON a.key = b.key; 

在MaxCompute Studio中调用,可以享受语法高亮和错误提示等功能。执行的调用代码如下:

image

执行的状态图如下:

image

放大执行过程仔细观察,图中可以发现几点有意思的地方:

image

上述执行输出的结果如下:

+------+---------+
| name | address |
+------+---------+
| 小明 | 杭州 |
+------+---------+

其他用法

经常有用户误用参数化视图,将参数化视图的参数当做是宏替换参数来使用。这里说明一下。参数化视图实际上是函数调用,而不是宏替换。如下面的例子:

CREATE VIEW paramed_view(@a TABLE(key STRING, value STRING), @b STRING) 
AS SELECT * FROM @a ORDER BY @b;

-- 调用示例
select * from paramed_view(src, 'key');

上面的例子中,用户的期望是 ORDER BY @b 被宏替换为 ORDER BY key,即根据输入参数,决定了按照key列做排序。然而,实际上参数@b是作为一个值来传递的,ORDER BY @b 相当于 ORDER BY 'key',即 ORDER BY一个字符串常量('key')而不是一列。要想实现"让调用者决定排序列"这一功能,可以考虑下述做法。

CREATE VIEW orderByFirstCol(@a TABLE(columnForOrder ANY, * ANY)) AS SELECT `(columnForOrder)?+.+` FROM (SELECT * FROM @a ORDER BY columnForOrder) t;

-- 调用示例
select * from orderByFirstCol((select key, * from src));

上面的例子,要求调用者将要排序的列放在第一列,于是在调用的时候使用子查询将src的需要排序的列抽取到最前面。视图返回的 (columnForOrder)?+.+ 是一个正则通配符,匹配columnForOrder之外的所有列,列表达式使用正则表达式可参考SELECT语法介绍>列表达式关于正则表达式的说明。

UDF:函数重载方式

问题

MaxCompute 的 UDF 使用重载 evalaute 方法的方式来重载函数,如下面的UDF定义了两个重载,当输入是 String 类型时,输出String类型,输入是BIGINT类型时,输出DOUBLE类型。

public UDFClass extends UDF {

    public String evaluate(String input) { return input + "123";  }

    public Double evaluate(Long input) { return input + 123.0; }
}

这种方式固然能解决一些问题,但有一定的局限性。比如不支持泛型,要做一个接受任何类型的函数,就必须为每种类型都写一个evaluate函数。有的时候重载甚至是不能实现的,比如ARRAY 和 ARRAY 的重载是做不到的。

public UDFClass extends UDF {

    public String evaluate(List<Long> input) { return input.size(); }

    // 这里会报错,因为在java类型擦除后,这个函数和 String evaluate(List<Long> input) 的参数是一样的
    public Double evaluate(List<Double> input) { input.size(); } 

    // UDF 不支持下面这种定义方式
    public String evaluate(List<Object> input) { return input.size(); }
}

PYTHON UDF 或 UDTF 在不提供 Resolve 注解(annotation)的时候,会根据参数个数决定输入参数,也支持变长,因此非常灵活。但也因为过于灵活,编译器无法静态找到某些错误。比如

class Substr(object):
    def evaluate(self, a, b):
      return a[b:];

上面的函数接受两个参数,从实现上看,第一个参数需要是STRING类型,第二个参数应该是整形。而这个限制需要用户在调用时自己去把握。即使用户传错了参数,编译器也没有办法报错。同时,这种方式定义的UDF返回值类型只能是STRING,不够灵活。

改进

要解决上面的问题。可以考虑使用UDT。 UDT经常被简单在调用JDK中的方法的时候使用,比如 java.util.Objects.toString(x) 将任何对象 x 转成STRING类型。但是在自定义函数方面同样也有很好的用途。 UDT支持泛型,支持类继承,支持变长等功能,让定义函数更方便。如下面的例子:

public class UDTClass {

    // 这个函数接受一个数值类型(可以是 TINYINT, SMALLINT, INT, BIGINT, FLOAT, DOUBLE 以及任何以Number为基类的UDT),返回DOUBLE
    public static Double doubleValue(Number input) {
        return input.doubleValue();
    }

    // 这个方法,接受一个数值类型参数和一个任意类型的参数,返回值类型与第二个参数的类型相同
    public static <T extends Number, R> R nullOrValue(T a, R b) {
        return a.doubleValue() > 0 ? b : null;
    }

    // 这个方法接受一个任意元素类型的array或List,返回BIGINT
    public static Long length(java.util.List<? extends Object> input) {
        return input.size();
    }

    // 注意这个在不做强制转换的情况下参数只能接受 UDT 的 java.util.Map<Object, Object> 对象。如果需要传入任何map对象,比如 map<bigint,bigint> 可以考虑:
    // 1. 定义函数时使用java.util.Map<? extends Object, ? extends Object>
    // 2. 调用时强转,比如 UDTClass.mapSize(cast(mapObj as java.util.Map<Object, Object>))
    public static Long mapSize(java.util.Map<Object, Object> input) {
        return input.size();
    }
}

UDT 能够提供灵活的函数定义方式。但是有的时候UDF 需要通过 com.aliyun.odps.udf.ExecutionContext(在setup方法中传入)来获取一些上下文。现在UDT也可以通过 com.aliyun.odps.udt.UDTExecutionContext.get() 方法来或者这样的一个 ExecutionContext 对象。

Aggregator 与 UDTF:Annotation方式

问题

MaxCompute 的 UDAF 和 UDTF 使用Resolve注解来决定函数Signature。比如下面的方式定义了一个UDTF,该UDTF接受一个BIGINT参数,返回DOUBLE类型。

@com.aliyun.odps.udf.annotation.Resolve("BIGINT->DOUBLE")
public class UDTFClass extends UDTF {
    ...
}

这种方式的局限性很明显,输入参数和输出参数都是固定的,没办法重载。

改进

MaxCompute对Resolve注解的语法做了许多扩展,现在能够支持一定的灵活性。

  • 参数列表中可以使用星号('*'),表示接受任意长度的,任意类型的输入参数。比如 @Resolve('double,*->String'),接受第一个是double,后接任意类型,任意个数的参数列表。这里需要UDF的作者在代码里面自己去判断输入的个数和类型,然后做出相应的动作(可以对比 C 语言里面的 printf 函数来理解)。注意星号用在返回值列表中时,表示的是不同的含义,在后续第三点中说明。
  • 参数列表中可以使用 ANY 关键字,表示任意类型的参数。比如 @Resolve('double,any->string'),接受第一个是double,第二个任意类型的参数列表。注意,ANY在返回值列表中不能使用,也不能在复杂类型的子类型中使用(如不能写ARRAY)。
  • UDTF的返回值可以使用星号,表示返回任意多个string类型。这里需要注意,返回值的个数并非真的是任意多个,而是与调用函数时给出的alias个数有关。比如@Resolve("ANY,ANY->DOUBLE,*"),调用方式是 UDTF(x, y) as (a, b, c),这里as后面给出了三个alias (a, b, c),编译器会认定a为double类型(annotation中返回值第一列的类型是给定的),b,c为string类型,而因为这里给出了三个返回值,所以UDTF在forward的时候,也一定要forward长度为3的数组,否则会出现运行时错误。注意这个错误是无法在编译时给出的,因此通常需要UDTF的作者与调用者互相沟通好,调用者在SQL中给出alias个数的时候,一定要按照UDTF的需要来写。由于Aggregator返回值个数固定是1,所以这个功能对UDAF无意义。

用一个例子来说明。如下UDTF:

import com.aliyun.odps.udf.UDFException;
import com.aliyun.odps.udf.UDTF;
import com.aliyun.odps.udf.annotation.Resolve;
import org.json.JSONException;
import org.json.JSONObject;

@Resolve("STRING,*->STRING,*")
public class JsonTuple extends UDTF {

  private Object[] result = null;

  @Override
  public void process(Object[] input) throws UDFException {
    if (result == null) {
      result = new Object[input.length];
    }

    try {
      JSONObject obj = new JSONObject((String)input[0]);
      for (int i = 1; i < input.length; i++) {
        // 返回值要求变长部分都是STRING
        result[i] = String.valueOf(obj.get((String)(input[i])));
      }
      result[0] = null;
    } catch (JSONException ex) {
      for (int i = 1; i < result.length; i++) {
        result[i] = null;
      }
      result[0] = ex.getMessage();
    }
    forward(result);
  }
}

这个UDTF的返回值个数会根据输入参数的个数来决定。输出参数的第一个是一个JSON文本,后面是需要从JSON中解析的key。返回值第一个是解析JSON过程中的出错信息,如果没有出错,则后续根据输入的key依次输出从json中解析出来的内容。使用示例如下。

-- 根据输入参数的个数定制输出alias个数
SELECT my_json_tuple(json, ’a‘, 'b') as exceptions, a, b FROM jsons;

-- 变长部分可以一列都没有
SELECT my_json_tuple(json) as exceptions, a, b FROM jsons;

-- 下面这个SQL会出现运行时错误,因为alias个数与实际输出个数不符
-- 注意编译时无法发现这个错误
SELECT my_json_tuple(json, 'a', 'b') as exceptions, a, b, c FROM jsons;

上面虽然做出了许多扩展,但是这些扩展并不一定能满足所有的需求。这时候依然可以考虑使用UDT。UDT也是可以用来实现Aggregator和UDTF的功能的。详细可以参考UDT示例文档,“聚合操作的实现示例” 及 “表值函数的实现示例” 的内容。

总结

MaxCompute自定义函数的函数原型不够灵活,在数据开发过程中带来诸多不便利,本文列举了各种函数定义方式存在的问题与解决方案,希望对大家有帮助,同时也告诉大家MaxCompute一直在努力为大家提供更好的服务。

相关实践学习
基于MaxCompute的热门话题分析
本实验围绕社交用户发布的文章做了详尽的分析,通过分析能得到用户群体年龄分布,性别分布,地理位置分布,以及热门话题的热度。
SaaS 模式云数据仓库必修课
本课程由阿里云开发者社区和阿里云大数据团队共同出品,是SaaS模式云原生数据仓库领导者MaxCompute核心课程。本课程由阿里云资深产品和技术专家们从概念到方法,从场景到实践,体系化的将阿里巴巴飞天大数据平台10多年的经过验证的方法与实践深入浅出的讲给开发者们。帮助大数据开发者快速了解并掌握SaaS模式的云原生的数据仓库,助力开发者学习了解先进的技术栈,并能在实际业务中敏捷的进行大数据分析,赋能企业业务。 通过本课程可以了解SaaS模式云原生数据仓库领导者MaxCompute核心功能及典型适用场景,可应用MaxCompute实现数仓搭建,快速进行大数据分析。适合大数据工程师、大数据分析师 大量数据需要处理、存储和管理,需要搭建数据仓库?学它! 没有足够人员和经验来运维大数据平台,不想自建IDC买机器,需要免运维的大数据平台?会SQL就等于会大数据?学它! 想知道大数据用得对不对,想用更少的钱得到持续演进的数仓能力?获得极致弹性的计算资源和更好的性能,以及持续保护数据安全的生产环境?学它! 想要获得灵活的分析能力,快速洞察数据规律特征?想要兼得数据湖的灵活性与数据仓库的成长性?学它! 出品人:阿里云大数据产品及研发团队专家 产品 MaxCompute 官网 https://www.aliyun.com/product/odps&nbsp;
目录
相关文章
|
2月前
|
SQL 消息中间件 分布式计算
大数据-124 - Flink State 01篇 状态原理和原理剖析:状态类型 执行分析
大数据-124 - Flink State 01篇 状态原理和原理剖析:状态类型 执行分析
83 5
|
13天前
|
SQL 分布式计算 DataWorks
DataWorks产品测评|基于DataWorks和MaxCompute产品组合实现用户画像分析
本文介绍了如何使用DataWorks和MaxCompute产品组合实现用户画像分析。首先,通过阿里云官网开通DataWorks服务并创建资源组,接着创建MaxCompute项目和数据源。随后,利用DataWorks的数据集成和数据开发模块,将业务数据同步至MaxCompute,并通过ODPS SQL完成用户画像的数据加工,最终将结果写入`ads_user_info_1d`表。文章详细记录了每一步的操作过程,包括任务开发、运行、运维操作和资源释放,帮助读者顺利完成用户画像分析。此外,还指出了文档中的一些不一致之处,并提供了相应的解决方法。
|
2月前
|
存储 分布式计算 NoSQL
大数据-40 Redis 类型集合 string list set sorted hash 指令列表 执行结果 附截图
大数据-40 Redis 类型集合 string list set sorted hash 指令列表 执行结果 附截图
29 3
|
2月前
|
消息中间件 分布式计算 NoSQL
大数据-41 Redis 类型集合(2) bitmap位操作 geohash空间计算 stream持久化消息队列 Z阶曲线 Base32编码
大数据-41 Redis 类型集合(2) bitmap位操作 geohash空间计算 stream持久化消息队列 Z阶曲线 Base32编码
30 2
|
3月前
|
JSON 数据可视化 数据挖掘
Polars函数合集大全:大数据分析的新利器
Polars函数合集大全:大数据分析的新利器
150 1
|
2月前
|
消息中间件 NoSQL Kafka
大数据-116 - Flink DataStream Sink 原理、概念、常见Sink类型 配置与使用 附带案例1:消费Kafka写到Redis
大数据-116 - Flink DataStream Sink 原理、概念、常见Sink类型 配置与使用 附带案例1:消费Kafka写到Redis
200 0
|
2月前
|
SQL 消息中间件 分布式计算
大数据-115 - Flink DataStream Transformation 多个函数方法 FlatMap Window Aggregations Reduce
大数据-115 - Flink DataStream Transformation 多个函数方法 FlatMap Window Aggregations Reduce
43 0
|
3月前
|
机器学习/深度学习 分布式计算 大数据
Spark 适合解决多种类型的大数据处理问题
【9月更文挑战第1天】Spark 适合解决多种类型的大数据处理问题
84 3
|
4月前
|
分布式计算 搜索推荐 物联网
大数据及AI典型场景实践问题之通过KafKa+OTS+MaxCompute完成物联网系统技术重构如何解决
大数据及AI典型场景实践问题之通过KafKa+OTS+MaxCompute完成物联网系统技术重构如何解决
|
4月前
|
人工智能 分布式计算 架构师
大数据及AI典型场景实践问题之基于MaxCompute构建Noxmobi全球化精准营销系统如何解决
大数据及AI典型场景实践问题之基于MaxCompute构建Noxmobi全球化精准营销系统如何解决

相关产品

  • 云原生大数据计算服务 MaxCompute