MaxCompute UDF配置汇总

本文涉及的产品
云原生大数据计算服务MaxCompute,500CU*H 100GB 3个月
简介: MaxCompute UDF为数据开发人员提供了较SQL更加灵活的数据处理方式,使得在数据处理过程中,不再只是局限在SQL提供的能力,可以根据自己的需求对功能进行定制,也可以引入其他优秀的程序包简化开发的难度和工作量。

MaxCompute UDF为数据开发人员提供了较SQL更加灵活的数据处理方式,使得在数据处理过程中,不再只是局限在SQL提供的能力,可以根据自己的需求对功能进行定制,也可以引入其他优秀的程序包简化开发的难度和工作量。有的时候一个UDF因为经过多轮迭代和长时间积累,可能功能会变的非常丰富同时又非常的复杂,常常需要利用配置进行控制,本文对几种通用的配置方案进行了汇总,供各位参考。


一、函数参数方式

执行示例:

select  udf1(param1, param2) from dual;
select udf1(param1,map("k","v")) from dual;

实现代码

public class udf1 extends UDF {

    public String evaluate(String a) {
        return "default value:" + a;
    }

    public String evaluate(String a, String b) {
        return "new value:" + a + "," + b;
    }

    public String evaluate(String a, String b, String c) {
        return "new value:" + a + "," + b + "," + c;
    }

    public String evaluate(String a, Integer b,List<String> c) {
        return "new value:" + c.get(b);
    }
    
    public String evaluate(String a,Map<String,String> map) {
        return "new value" + map.get(a);
    }
    
    public String evaluate(String a,Struct struct) {   
        return "new value" + struct.getFieldValue(a);
    }
}

特殊说明:
1,在本例最后一个实现中的Struct类型,为com.aliyun.odps.data.Struct
2,更多复杂参数类型,可以参考https://yq.aliyun.com/articles/225026


二、执行参数方式

注册函数

create function udf2 as com.aliyun.udf. udf2 using "udf-1.0-SNAPSHOT.jar";

执行示例

set myconfig.config.a=hello;
set myconfig.config.other=2;
select udf2(param1) from dual;

实现代码

public class udf2 extends UDF {
    // TODO define parameters and return type, e.g:  public String evaluate(String a, String b)
    private String flags;
    public void setup(ExecutionContext ctx) throws UDFException {

        Properties ps= ctx.getConfigurations();
        flags = ps.getProperty("myconfig.config.a");
        
       /* 
       说明,目前ctx.getConfigurations();仅能kv返回,不能支持便利操作,所以如下代码并不能返回全部变量
       System.out.println("keys count : "+ps.size());

        Set<Map.Entry<Object, Object>> ent = ps.entrySet() ;
        Iterator<Map.Entry<Object, Object>> it = ent.iterator();
        while(((Iterator) it).hasNext()) {
            Map.Entry<Object,Object> kv = it.next();
            System.out.println("key : "+kv.getKey() + " value : " + kv.getValue());
        }
        */
    }
    public String evaluate(String s) {
        if (flags == null) {
            return "default value:" + s;
        } else {
            return "new value:" + flags + s;
        }
    }
}

特殊说明:
1,set 的kv对在getProperty的时候以“等号”区分,所以内容中不能再包含等号
2,set 设置的变量会在当前session中一直保留,设置一次后续的多个sql都能读取到,要注意范围


三、配置文件方式

注册函数

add file ./cfg.txt;
create function udf3 as com.aliyun.udf.udf3 using "udf-1.0-SNAPSHOT.jar, cfg.txt";

执行示例

select udf3(param1) from dual;

实现代码

 public class udf3 extends UDF {
 private String flags;
 public void setup(ExecutionContext ctx) throws UDFException {
     try {
         InputStream in = ctx.readResourceFileAsStream("cfg.txt");
         BufferedReader br = new BufferedReader(new InputStreamReader(in));
         flags = br.readLine();
     }catch (IOException e) {
         System.out.println("read error, "+e.toString());
         throw new UDFException(e);
     }

 }

 public String evaluate(String s) {
     if (flags == null) {
         return "default value:"+ s;
     }
     else{
         return "new value:" + flags + s;
     }
 }
}

特殊说明:
1,要使用resource配置文件,必须要在注册函数的时候using相应的resource,才能被UDF发现
2,除了使用resource文件外,其实还可以使用odps table,与本例相似,所以不再说明
3,函数注册完成后,对resource配置文件进行跟新覆盖,即可更新配置


以上三种方式基本上可以覆盖绝大多数情况,除此之外还可以组合使用,例如:某个功能复杂的特征工程的UDF,通过第一种和第二种方式,都需要在执行的时候配置多个参数,那么不如把他写成resource配置文件,同时组内开发人员众多,每个人都想使用自己的配置文件,另外又想根据情况随时做变量的微调;那么就可以将如上三种方案融合,变成一种综合方案,示例如下:

注册函数:

 add file usr1.txt;
 add file usr2.txt;
 add file usr3.txt;
 add file usr4.txt;
 create function udf4 as com.aliyun.udf.udf4 using "udf-1.0-SNAPSHOT.jar, usr1.txt,usr2.txt, usr3.txt,usr4.txt";

执行示例

select udf4(param1, map()) from dual;  -- 默认分支

set myconfig.config.file=usr3.txt;
select udf4(param1,map()) from dual;  --有配置文件的分支

set myconfig.config.file=usr3.txt;
select udf4(param1, map("k","v")) from dual; --有配置文件,但是通过函数参数调整配置的分支

实现代码

public class udf4 extends UDF {
    private String flags;
    public void setup(ExecutionContext ctx) throws UDFException {
        try {
            Properties ps= ctx.getConfigurations();
            String path = ps.getProperty("myconfig.config.file");

            if (path == null || path.length() <= 0) return ;

            InputStream in = ctx.readResourceFileAsStream(path);
            BufferedReader br = new BufferedReader(new InputStreamReader(in));
            flags = br.readLine();
        }catch (IOException e) {
            System.out.println("read error, "+e.toString());
            throw new UDFException(e);
        }

    }

    public String evaluate(String s, Map<String,String> mflags) {
        if (mflags.size() > 0) {
            return "new value 1:" + s;
        }
        if (flags == null) {
            return "default value:"+ s;
        }
        else{
            return "new value 2:" + flags + s;
        }
    }
}

特殊说明
1,在注册函数的时候,必须将所有可能用到的配置文件都注册,不然函数无法注册成功

除了普通的UDF外,UDAF、UDTF可用同样的方法进行配置。Python UDF的处理方式可参见:https://help.aliyun.com/document_detail/73359.html?spm=a1z2e.8101737.webpage.dtitle2.bb2a6a6c6Di3vt

相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
基于MaxCompute的热门话题分析
Apsara Clouder大数据专项技能认证配套课程:基于MaxCompute的热门话题分析
目录
相关文章
|
3月前
|
资源调度 安全 Java
Java 大数据在智能教育在线实验室设备管理与实验资源优化配置中的应用实践
本文探讨Java大数据技术在智能教育在线实验室设备管理与资源优化中的应用。通过统一接入异构设备、构建四层实时处理管道及安全防护双体系,显著提升设备利用率与实验效率。某“双一流”高校实践显示,设备利用率从41%升至89%,等待时间缩短78%。该方案降低管理成本,为教育数字化转型提供技术支持。
84 1
|
2月前
|
存储 搜索推荐 算法
Java 大视界 -- Java 大数据在智能金融理财产品风险评估与个性化配置中的应用(195)
本文深入探讨了Java大数据技术在智能金融理财产品风险评估与个性化配置中的关键应用。通过高效的数据采集、存储与分析,Java大数据技术助力金融机构实现精准风险评估与个性化推荐,提升投资收益并降低风险。
Java 大视界 -- Java 大数据在智能金融理财产品风险评估与个性化配置中的应用(195)
|
11月前
|
分布式计算 资源调度 大数据
大数据-110 Flink 安装部署 下载解压配置 Standalone模式启动 打包依赖(一)
大数据-110 Flink 安装部署 下载解压配置 Standalone模式启动 打包依赖(一)
286 0
|
11月前
|
存储 分布式计算 druid
大数据-152 Apache Druid 集群模式 配置启动【下篇】 超详细!(一)
大数据-152 Apache Druid 集群模式 配置启动【下篇】 超详细!(一)
131 1
大数据-152 Apache Druid 集群模式 配置启动【下篇】 超详细!(一)
|
11月前
|
运维 监控 数据可视化
大数据-171 Elasticsearch ES-Head 与 Kibana 配置 使用 测试
大数据-171 Elasticsearch ES-Head 与 Kibana 配置 使用 测试
377 1
|
11月前
|
消息中间件 分布式计算 druid
大数据-152 Apache Druid 集群模式 配置启动【下篇】 超详细!(二)
大数据-152 Apache Druid 集群模式 配置启动【下篇】 超详细!(二)
115 2
|
11月前
|
存储 消息中间件 druid
大数据-151 Apache Druid 集群模式 配置启动【上篇】 超详细!
大数据-151 Apache Druid 集群模式 配置启动【上篇】 超详细!
207 1
|
11月前
|
SQL 消息中间件 分布式计算
大数据-141 - ClickHouse 集群 副本和分片 Zk 的配置 Replicated MergeTree原理详解(一)
大数据-141 - ClickHouse 集群 副本和分片 Zk 的配置 Replicated MergeTree原理详解(一)
224 0
|
11月前
|
SQL 大数据
大数据-141 - ClickHouse 集群 副本和分片 Zk 的配置 Replicated MergeTree原理详解(二)
大数据-141 - ClickHouse 集群 副本和分片 Zk 的配置 Replicated MergeTree原理详解(二)
246 0
|
11月前
|
消息中间件 NoSQL Kafka
大数据-116 - Flink DataStream Sink 原理、概念、常见Sink类型 配置与使用 附带案例1:消费Kafka写到Redis
大数据-116 - Flink DataStream Sink 原理、概念、常见Sink类型 配置与使用 附带案例1:消费Kafka写到Redis
619 0

热门文章

最新文章