00:00
我们已经知道了系统函数是liq当中已经底层内嵌实现的。能够满足我们特定功能的一大类函数,它的个数是非常多的,但是也不可能覆盖我们所有的功能,所以如果遇到系统函数不支持的需求,这个时候我们应该怎么做呢?诶,那当然了,就可以使用自定义函数,就是所谓的user DeFine function。UD来进行实现啊,那事实上现在我们的这个系统内置数啊,仍在不断的扩充,在1.3这个版本当中,这部分其实还不是完全稳定的一个状态,如果我们现在认为自己开发出来的自定义函数uf足够通用的话,那么其实也可以在项目跟踪工具J上面向这个开发团队啊提出一选,然后。请求将我们开发的函数添加到下一个版本的系统函数当中去啊,所以这个也是一个动态变化的过程,自定义函数也可以在下一个版本当中变成系统函数。
01:07
那当前flink table API和CQ提供了各种各样不U的就是,所以我们如果想要实现对应的一个udf的话,那相当于就是继承对应的抽类,实现里边的抽象方法,然后呃,定义对应的对象就可以使用了。当前的udf呢,主要有。以下这么几大类,我们可以看到,首先标量函数啊,那还有聚合函数,这跟我们系统函数里边的两大类是一样的。它们的定义本身也是完全相同,那标量函数就是将输入的标量值转换成一个新的值,标量值啊。聚合函数呢,就是将多行数据里边的标样值转换成一个新的值,那另外多了两种,一类叫表函数,另外一类叫表聚合函数,啊其实表函数之前我们都已经接触过,什么叫表函数呢?诶,那之前我们讲到窗口的时候,新版本的窗口我们说使用的是窗口TF用的是窗口函数。
02:17
它的调用过程其实就是。把当前的窗口转换成了一个新的表,基于之前的数据表,然后结合这个窗口的信息转换成了新的一个表。所以所谓的表函数就是。把当前的标量值转换成了一个或多个新的行数据,也就是说扩展成了一个新的表,它的输出就不再是一个标量值了,而是一张表或者说是多行数据。之前我们这个标量函数和聚合函数可以认为是输出一行数据,而表函数输出多行数。
03:01
那同样什么叫做表距和函数呢?表距和函数就是把多行数据里的标量值转换成一个或多个新的行数据,也就是说输出一张表,它和表函数一样,同样是做了一个。输出表的这样的一个操作,而同样它又结合了聚合函数的特点,它是要把多行数据里边的标量值进行计算转换,哎,那所以我们可以把表聚合函数看成是表函数和聚合函数的一个结合体。那接下来我们就来具体看一下在代码当中怎么样去使用udf。呃,我们首先看一下整体的调用流程,如果想要在代码当中使用自定义的函数的话,那么首先我们需要实现对应的U点抽象类啊,那然后呢,实现了这个抽象类之后,还要创建它的对象,然后在表环境里边注册当前这个函数,接下来就可以调用了。
04:04
啊,那我们首先这个实现UD类,当然跟我们具体要实现的不同类关,这个放到具体讲解,那现在假如说我们已经实现了对应的uf,接下来首先就要注册函数了。注册函数的时候调用的是table en的create temporary function这样一个方法,那这个方法传入的呢?有两个参数,一个是当前要注册的函数名,另外一个是。实现了的udf类的class对象啊,那所以当前我们经过这样一个调用,就创建了一个临时的系统函数。比如说在这样一句。注册函数的这个代码里边,我们就注册了一个叫做my function的。系统函数,而这个系统函数对应呢,有一个实现的udf类啊,那就是在Java代码里边有一个类叫做my function,我们可能需要让他去抽象的,比方说标量函数,或者说聚合函数,或者表函数。
05:13
有了这个东西之后,诶,这里我们看到这个create temporary system function,它是创建一个所谓的系统函数,那这个什么叫系统函数呢?也就是说当前我们创建之后,可以把它当成系统函数来使用了,它的作用域是全局的。啊,那我们能不能不让它的作用扩展到全局呢?也是可以的,就是如果这里我们调用的方法不调create temporary function,而是调一个create temporary function方法的话。那么我们直接注册的函数呢?是放在了当前的数据库和当前的catallo下啊,所以这就相当于不是系统函数了,因为只有在当前的目录,当前的数据库下边才能够去调用,所以有时候会对应于系统函数,把它叫做catalog方式,也就是目录函式。
06:06
它的完整名称前面也是有catalog和database数据库的。那所以当前这个注册的函数就相当于跟我们的表作用域是完全一样的,那一般情况我们其实不需要这么麻烦,临时我们一般在代码里边也都是临时使用嘛,所以直接调用create system function,把它注册成一个系统函数,然后直接使用就可以了。那注册之后怎么使用呢?有两种方式,一种是使用API去调用,我们可以先看一下使用table API的这种调用方式啊,它是当前的表环境,然后from my table,哎,那我们知道这种方式就是首先要获取到当前的。表的对象啊,那就是我们在表环境里面注册过my table,现在呢,From方法把这个对应的table对象先获取到,然后接下来调用table的点select,这是叫做查询转换了,查询的过程当中我们要提取什么呢?比方说我现在要针对MY的这个字段调用my function,把它进行转换,然后提取。
07:12
嗯,那这个my function函数怎么调用呢?使用方法来调用自定义的U点。这里的靠方法两个参数,一个就是我们注册好的函数名称,注意这个必须在之前就已经调用这个方法注册过,否则如果只有这里的类实现好的udf类的话,这里是没有办法判断的,必须要注册,然后另外呢,呃,就是传入的参数,也就是调用当前my function的时候,哎,我们要把哪个字段传进来,然后进行转换进行计算呢?诶,那这里边就是我们想要传入的对应的那个字段。所以在table API里边最常见的调用方式就是这样啊,那当然了,还有另外一种方式,就是我们也可以直接就是在这里,我们不在表环境里边注册当前的MY方式,而是使用一个叫做内联inline的方式去调用udf,诶,那就是前面我们没有注册,没有注册怎么办呢?没关系,直接把当前的。
08:14
自己定义好的那个my function啊,对应的那一个类名,或者说这里我们定义的那个叫做substre function啊对应的这个类名,然后点class,把当前的这个类。传进来,然后接下来还是同样把对应要传入的参数字段传入调用方法啊,这两种方式都是可以的。但是我们会发现这种调用方式稍微的有一点麻烦,这里边我们又要使用一个call,然后还得考虑到就是各种注册啊,使用的这个过程并不符合我们平常的习惯啊,那所以table API,其实而而且我们说这个table API本身它里边的系统也比较少,功能是比较受局限的,所以这种方式其实在实际项目当中应用比较少,那更常见的方法。
09:06
当然就是直接在CQ当中调用函数了,因为之前我们不是已经把它注册为系统函数了吗?那接下来直接就可以写一句CQ啊,那table EV去执行q que进行一个CQ查询,里边写的就是select my function field啊,这是我们注册好的函数,然后这是我们想要提取的字段,把它传进来,这就跟我们在标准CQ里边使用系统函数的方法是完全一致的。那所以后面呢,我们就主要针对CQ的这种用法,来对不同类型的udf做一个详细的讲解和介绍。
我来说两句