[1.2.0新功能系列:三]Apache doris 1.2.0 Java UDF 函数开发及使用

本文涉及的产品
密钥管理服务KMS,1000个密钥,100个凭据,1个月
简介: [1.2.0新功能系列:三]Apache doris 1.2.0 Java UDF 函数开发及使用

概述


我们在使用各个SQL引擎时,会有纷繁复杂的查询需求。一部分可以通过引擎自带的内置函数去解决,但内置函数不可能解决所有人的问题,所以一般SQL引擎会提供UDF功能,方便用户通过自己写逻辑来满足特定的需求,Doris也不例外。


在java UDF之前,Doris提供了两种用户可以自己实现UDF的方式:


远程UDF,其优缺点如下:


支持通过 RPC 的方式访问用户提供的 UDF Service,以实现用户自定义函数的执行


只要支持Protobuf的各类语言都能使用,有足够的安全和灵活性


额外的网络开销和基于protobuf的开发模式让该使用方式的用户望而却步

原生UDF,其优缺点如下:


支持使用C++编写UDF,执行效率高、速度快


跟Doris代码耦合度高,需要自己打包编译Doris源码


只支持C++语言并且容易造成BE挂掉


熟悉大数据组件(Hive Spark等)的用户有一定的门槛


看起来上述UDF的两种方式实现起来有点复杂。有没有相对简单,门槛较低,跟Doris代码耦合度低,对Java友好的UDF方式呢?


在 Doris 1.2.0 版本我们正式支持 Java UDF 函数,你可以像之前写 Hive udf函数一样去写自己的Doris udf函数来处理自己复杂的业务逻辑。


SinceVersion 1.2.0


Java UDF 为用户提供UDF编写的Java接口,以方便用户使用Java语言进行自定义函数的执行。相比于 Native 的 UDF 实现,Java UDF 有如下优势和限制:


优势


兼容性:使用Java UDF可以兼容不同的Doris版本,所以在进行Doris版本升级时,Java UDF不需要进行额外的迁移操作。与此同时,Java UDF同样遵循了和Hive/Spark等引擎同样的编程规范,使得用户可以直接将Hive/Spark的UDF jar包迁移至Doris使用。


安全:Java UDF 执行失败或崩溃仅会导致JVM报错,而不会导致 Doris 进程崩溃。


灵活:Java UDF 中用户通过把第三方依赖打进用户jar包,而不需要额外处理引入的三方库。


使用限制


性能:相比于 Native UDF,Java UDF会带来额外的JNI开销,不过通过批式执行的方式,我们已经尽可能的将JNI开销降到最低。


向量化引擎:Java UDF当前只支持向量化引擎。


doris 提供


UDF:用户自定义函数,user defined function。一对一的输入输出,(最常用的)。


UDAF:用户自定义聚合函数。user defined aggregate function,多对一的输入输出,类似 count sum max 等统计函数


怎么实现 Doris Java UDF函数


下面我们来开始讲解怎么编写和使用 doris java udf函数。


Doris java udf 函数是基于 Hive udf 框架来实现的


继承org.apache.hadoop.hive.ql.exec.UDF


重写evaluate(),


特殊说明:


evaluate()方法不是由接口定义的,因为它可接受的参数个数,数据类型都是不确定的。Doris 会检查UDF, 看能否找到和函数调用相匹配的evaluate()方法


这里演示的是我们怎么实现一个 AES 加解密的函数


函数开发


我们创建一个普通的java maven 工程


pom.xml依赖如下:

<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 <modelVersion>4.0.0</modelVersion>
 <groupId>org.apache.doris</groupId>
 <artifactId>doris.java.udf.demo</artifactId>
 <version>1.0-SNAPSHOT</version>
 <packaging>jar</packaging>
 <name>doris.java.udf.demo</name>
 <url>http://maven.apache.org</url>
 <properties>
   <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
 </properties>
 <dependencies>
   <dependency>
     <groupId>org.apache.hive</groupId>
     <artifactId>hive-exec</artifactId>
     <version>2.3.5</version>
   </dependency>
 </dependencies>
 <build>
   <finalName>java-udf-demo</finalName>
   <plugins>
     <plugin>
       <groupId>org.apache.maven.plugins</groupId>
       <artifactId>maven-jar-plugin</artifactId>
       <version>3.2.2</version>
     </plugin>
     <plugin>
       <groupId>org.apache.maven.plugins</groupId>
       <artifactId>maven-assembly-plugin</artifactId>
       <version>3.3.0</version>
       <configuration>
         <descriptorRefs>
           <descriptorRef>jar-with-dependencies</descriptorRef>
         </descriptorRefs>
       </configuration>
       <executions>
         <execution>
           <phase>package</phase>
           <goals>
             <goal>single</goal>
           </goals>
         </execution>
       </executions>
     </plugin>
     <plugin>
       <groupId>org.apache.maven.plugins</groupId>
       <artifactId>maven-compiler-plugin</artifactId>
       <configuration>
         <source>8</source>
         <target>8</target>
       </configuration>
     </plugin>
   </plugins>
 </build>
</project>

加解密工具类:

package org.apache.doris.udf.demo;
import javax.crypto.*;
import javax.crypto.spec.SecretKeySpec;
import org.apache.commons.lang3.StringUtils;
import java.security.SecureRandom;
/**
* AES encryption and decryption tool class
*
* @author zhangfeng
*/
public class AESUtil {
   private static final String defaultCharset = "UTF-8";
   private static final String KEY_AES = "AES";
   /**
    * AES encryption function method
    *
    * @param content
    * @param secret
    * @return
    */
   public static String encrypt(String content, String secret) {
       return doAES(content, secret, Cipher.ENCRYPT_MODE);
  }
   /**
    * AES decryption function method
    *
    * @param content
    * @param secret
    * @return
    */
   public static String decrypt(String content, String secret) {
       return doAES(content, secret, Cipher.DECRYPT_MODE);
  }
   /**
    * encryption and decryption
    *
    * @param content
    * @param secret
    * @param mode
    * @return
    */
   private static String doAES(String content, String secret, int mode) {
       try {
           if (StringUtils.isBlank(content) || StringUtils.isBlank(secret)) {
               return null;
          }
           //Determine whether to encrypt or decrypt
           boolean encrypt = mode == Cipher.ENCRYPT_MODE;
           byte[] data;
           //1.Construct a key generator, specified as the AES algorithm, case-insensitive
           KeyGenerator kgen = KeyGenerator.getInstance(KEY_AES);
           SecureRandom secureRandom = SecureRandom.getInstance("SHA1PRNG");
           //2. Initialize the key generator according to the ecnodeRules rules
           //Generate a 128-bit random source, based on the incoming byte array
           secureRandom.setSeed(secret.getBytes());
           //Generate a 128-bit random source, based on the incoming byte array
           kgen.init(128, secureRandom);
           //3.generate the original symmetric key
           SecretKey secretKey = kgen.generateKey();
           //4.Get the byte array of the original symmetric key
           byte[] enCodeFormat = secretKey.getEncoded();
           //5.Generate AES key from byte array
           SecretKeySpec keySpec = new SecretKeySpec(enCodeFormat, KEY_AES);
           //6.According to the specified algorithm AES self-generated cipher
           Cipher cipher = Cipher.getInstance(KEY_AES);
           //7.Initialize the cipher, the first parameter is encryption (Encrypt_mode) or decryption (Decrypt_mode) operation,
           // the second parameter is the KEY used
           cipher.init(mode, keySpec);
           if (encrypt) {
               data = content.getBytes(defaultCharset);
          } else {
               data = parseHexStr2Byte(content);
          }
           byte[] result = cipher.doFinal(data);
           if (encrypt) {
               //convert binary to hexadecimal
               return parseByte2HexStr(result);
          } else {
               return new String(result, defaultCharset);
          }
      } catch (Exception e) {
           System.out.println(e.getMessage());
      }
       return null;
  }
   /**
    * convert binary to hexadecimal
    *
    * @param buf
    * @return
    */
   public static String parseByte2HexStr(byte buf[]) {
       StringBuilder sb = new StringBuilder();
       for (int i = 0; i < buf.length; i++) {
           String hex = Integer.toHexString(buf[i] & 0xFF);
           if (hex.length() == 1) {
               hex = '0' + hex;
          }
           sb.append(hex.toUpperCase());
      }
       return sb.toString();
  }
   /**
    * Convert hexadecimal to binary
    *
    * @param hexStr
    * @return
    */
   public static byte[] parseHexStr2Byte(String hexStr) {
       if (hexStr.length() < 1) {
           return null;
      }
       byte[] result = new byte[hexStr.length() / 2];
       for (int i = 0; i < hexStr.length() / 2; i++) {
           int high = Integer.parseInt(hexStr.substring(i * 2, i * 2 + 1), 16);
           int low = Integer.parseInt(hexStr.substring(i * 2 + 1, i * 2 + 2), 16);
           result[i] = (byte) (high * 16 + low);
      }
       return result;
  }
}

加密函数

package org.apache.doris.udf.demo;
import org.apache.hadoop.hive.ql.exec.UDF;
import org.apache.commons.lang3.StringUtils;
public class AESEncrypt extends UDF {
   public String evaluate(String content, String secret) throws Exception {
       if (StringUtils.isBlank(content)) {
           throw new Exception("content not is null");
      }
       if (StringUtils.isBlank(secret)) {
           throw new Exception("Secret not is null");
      }
       return AESUtil.encrypt(content, secret);
  }
}

解密函数

package org.apache.doris.udf.demo;
import org.apache.hadoop.hive.ql.exec.UDF;
import org.apache.commons.lang3.StringUtils;
public class AESDecrypt extends UDF {
   public String evaluate(String content, String secret) throws Exception {
       if (StringUtils.isBlank(content)) {
           throw new Exception("content not is null");
      }
       if (StringUtils.isBlank(secret)) {
           throw new Exception("Secret not is null");
      }
       return AESUtil.decrypt(content, secret);
  }
}

函数打包

mvn clean package

这个时候我们可以得到一个 java-udf-demo.jar


注册函数


注册加密函数


这里有两个参数,一个是加密内容,一个是秘钥,返回值是一个字符串

CREATE FUNCTION ase_encryp(string,string) RETURNS string PROPERTIES (
   "file"="file:///Users/zhangfeng/work/doris.java.udf.demo/target/java-udf-demo.jar",
   "symbol"="org.apache.doris.udf.demo.AESEncrypt",
   "always_nullable"="true",
   "type"="JAVA_UDF"
);

注意:


这里我是单机测试,使用的是本地文件方式,如果你也是要本地文件方式需要再所有的 FE 及 BE 上相同目录下都要有这个文件


我们也可以使用http方式,让每个节点自己下载这个文件,我们更推荐这种方式,下面也给出这种方式的示例


Http 方式示例:

CREATE FUNCTION ase_encryp(string,string) RETURNS string PROPERTIES (
   "file"="http://192.168.31.54/work/doris.java.udf.demo/target/java-udf-demo.jar",
   "symbol"="org.apache.doris.udf.demo.AESEncrypt",
   "always_nullable"="true",
   "type"="JAVA_UDF"
);

然后我们执行我们刚才创建的函数


要加密的内容是:zhangfeng,秘钥是: java_udf_function

select ase_encryp('zhangfeng','java_udf_function');

从下图可以看到我们得到了加密后的结果

40.png注册解密函数

CREATE FUNCTION ase_decryp(string,string) RETURNS string PROPERTIES (
  "file"="file:///Users/zhangfeng/work/doris.java.udf.demo/target/java-udf-demo.jar",
  "symbol"="org.apache.doris.udf.demo.AESDecrypt",
  "always_nullable"="true",
  "type"="JAVA_UDF"
);

http方式:

CREATE FUNCTION ase_decryp(string,string) RETURNS string PROPERTIES (
  "file"="http://192.168.63.32/work/doris.java.udf.demo/target/java-udf-demo.jar",
  "symbol"="org.apache.doris.udf.demo.AESDecrypt",
  "always_nullable"="true",
  "type"="JAVA_UDF"
);

验证函数


我们对上面解密的结果进行解密操作

select ase_decryp('4442106BB8C98E74D19CEC0413467810','java_udf_function');

可以看到我们得到了正确的解密结果

41.png

目录
相关文章
|
2天前
|
存储 SQL Apache
Apache Doris 开源最顶级基于MPP架构的高性能实时分析数据库
Apache Doris 是一个基于 MPP 架构的高性能实时分析数据库,以其极高的速度和易用性著称。它支持高并发点查询和复杂分析场景,适用于报表分析、即席查询、数据仓库和数据湖查询加速等。最新发布的 2.0.2 版本在性能、稳定性和多租户支持方面有显著提升。社区活跃,已广泛应用于电商、广告、用户行为分析等领域。
Apache Doris 开源最顶级基于MPP架构的高性能实时分析数据库
|
3天前
|
监控 Cloud Native BI
8+ 典型分析场景,25+ 标杆案例,Apache Doris 和 SelectDB 精选案例集(2024版)电子版上线
飞轮科技正式推出 Apache Doris 和 SelectDB 精选案例集 ——《走向现代化的数据仓库(2024 版)》,汇聚了来自各行各业的成功案例与实践经验。该书以行业为划分标准,辅以使用场景标签,旨在为读者提供一个高度整合、全面涵盖、分类清晰且易于查阅的学习资源库。
|
3天前
|
SQL DataWorks 关系型数据库
阿里云 DataWorks 正式支持 SelectDB & Apache Doris 数据源,实现 MySQL 整库实时同步
阿里云数据库 SelectDB 版是阿里云与飞轮科技联合基于 Apache Doris 内核打造的现代化数据仓库,支持大规模实时数据上的极速查询分析。通过实时、统一、弹性、开放的核心能力,能够为企业提供高性价比、简单易用、安全稳定、低成本的实时大数据分析支持。SelectDB 具备世界领先的实时分析能力,能够实现秒级的数据实时导入与同步,在宽表、复杂多表关联、高并发点查等不同场景下,提供超越一众国际知名的同类产品的优秀性能,多次登顶 ClickBench 全球数据库分析性能排行榜。
|
9天前
|
开发框架 JavaScript 前端开发
HarmonyOS UI开发:掌握ArkUI(包括Java UI和JS UI)进行界面开发
【10月更文挑战第22天】随着科技发展,操作系统呈现多元化趋势。华为推出的HarmonyOS以其全场景、多设备特性备受关注。本文介绍HarmonyOS的UI开发框架ArkUI,探讨Java UI和JS UI两种开发方式。Java UI适合复杂界面开发,性能较高;JS UI适合快速开发简单界面,跨平台性好。掌握ArkUI可高效打造符合用户需求的界面。
47 8
|
4天前
|
SQL Java 程序员
倍增 Java 程序员的开发效率
应用计算困境:Java 作为主流开发语言,在数据处理方面存在复杂度高的问题,而 SQL 虽然简洁但受限于数据库架构。SPL(Structured Process Language)是一种纯 Java 开发的数据处理语言,结合了 Java 的架构灵活性和 SQL 的简洁性。SPL 提供简洁的语法、完善的计算能力、高效的 IDE、大数据支持、与 Java 应用无缝集成以及开放性和热切换特性,能够大幅提升开发效率和性能。
|
5天前
|
存储 Java 关系型数据库
在Java开发中,数据库连接是应用与数据交互的关键环节。本文通过案例分析,深入探讨Java连接池的原理与最佳实践
在Java开发中,数据库连接是应用与数据交互的关键环节。本文通过案例分析,深入探讨Java连接池的原理与最佳实践,包括连接创建、分配、复用和释放等操作,并通过电商应用实例展示了如何选择合适的连接池库(如HikariCP)和配置参数,实现高效、稳定的数据库连接管理。
15 2
|
5天前
|
监控 Java 数据库连接
在Java开发中,数据库连接管理是关键问题之一
在Java开发中,数据库连接管理是关键问题之一。本文介绍了连接池技术如何通过预创建和管理数据库连接,提高数据库操作的性能和稳定性,减少资源消耗,并简化连接管理。通过示例代码展示了HikariCP连接池的实际应用。
10 1
WK
|
4天前
|
开发框架 移动开发 Java
C++和Java哪个更适合开发移动应用
本文对比了C++和Java在移动应用开发中的优劣,从市场需求、学习难度、开发效率、跨平台性和应用领域等方面进行了详细分析。Java在Android开发中占据优势,而C++则适合对性能要求较高的场景。选择应根据具体需求和个人偏好综合考虑。
WK
11 0
WK
|
5天前
|
安全 Java 编译器
C++和Java哪个更适合开发web网站
在Web开发领域,C++和Java各具优势。C++以其高性能、低级控制和跨平台性著称,适用于需要高吞吐量和低延迟的场景,如实时交易系统和在线游戏服务器。Java则凭借其跨平台性、丰富的生态系统和强大的安全性,广泛应用于企业级Web开发,如企业管理系统和电子商务平台。选择时需根据项目需求和技术储备综合考虑。
WK
9 0
|
17天前
|
SQL Java API
Apache Flink 2.0-preview released
Apache Flink 社区正积极筹备 Flink 2.0 的发布,这是自 Flink 1.0 发布以来的首个重大更新。Flink 2.0 将引入多项激动人心的功能和改进,包括存算分离状态管理、物化表、批作业自适应执行等,同时也包含了一些不兼容的变更。目前提供的预览版旨在让用户提前尝试新功能并收集反馈,但不建议在生产环境中使用。
465 13
Apache Flink 2.0-preview released

热门文章

最新文章

推荐镜像

更多