[1.2.0新功能系列：三]Apache doris 1.2.0
概述
我们在使用各个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');
从下图可以看到我们得到了加密后的结果
image.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');
可以看到我们得到了正确的解密结果
image.png
