13.3、Flink Table API & SQL - Hive 之 Hive函数

我就是我 2024-04-17 06:06 69阅读 0赞

### 用户自定义函数 ###

用户可以在Flink中使用他们现有的Hive用户定义函数。

支持的UDF函数包括以下：

*  UDF
 *  GenericUDF
 *  GenericUDTF
 *  UDAF
 *  GenericUDAFResolver2

通过查询计划和执行，Hive的UDF和GenericUDF被自动转换为Flink的ScalarFunction, Hive的GenericUDTF被自动转换为Flink的TableFunction, Hive的UDAF和GenericUDAFResolver2被转换为Flink的AggregateFunction。

**使用Hive用户自定义函数，用户必须：**

*  将由Hive Metastore支持的HiveCatalog设置为会话的当前目录，其中包含该函数（详见：[Flink Table API & SQL - Catalogs（使用概览和样例）][Flink Table API _ SQL - Catalogs]）
 *  在Flink的类路径中包含一个包含该函数的jar
 *  使用Blink planner

### 使用用户自定义函数 ###

假设我们在Hive Metastore中注册了以下Hive函数：

/**
     * Test simple udf. Registered under name 'myudf'
     */
    public class TestHiveSimpleUDF extends UDF {
    
    	public IntWritable evaluate(IntWritable i) {
    		return new IntWritable(i.get());
    	}
    
    	public Text evaluate(Text text) {
    		return new Text(text.toString());
    	}
    }
    
    /**
     * Test generic udf. Registered under name 'mygenericudf'
     */
    public class TestHiveGenericUDF extends GenericUDF {
    
    	@Override
    	public ObjectInspector initialize(ObjectInspector[] arguments) throws UDFArgumentException {
    		checkArgument(arguments.length == 2);
    
    		checkArgument(arguments[1] instanceof ConstantObjectInspector);
    		Object constant = ((ConstantObjectInspector) arguments[1]).getWritableConstantValue();
    		checkArgument(constant instanceof IntWritable);
    		checkArgument(((IntWritable) constant).get() == 1);
    
    		if (arguments[0] instanceof IntObjectInspector ||
    				arguments[0] instanceof StringObjectInspector) {
    			return arguments[0];
    		} else {
    			throw new RuntimeException("Not support argument: " + arguments[0]);
    		}
    	}
    
    	@Override
    	public Object evaluate(DeferredObject[] arguments) throws HiveException {
    		return arguments[0].get();
    	}
    
    	@Override
    	public String getDisplayString(String[] children) {
    		return "TestHiveGenericUDF";
    	}
    }
    
    /**
     * Test split udtf. Registered under name 'mygenericudtf'
     */
    public class TestHiveUDTF extends GenericUDTF {
    
    	@Override
    	public StructObjectInspector initialize(ObjectInspector[] argOIs) throws UDFArgumentException {
    		checkArgument(argOIs.length == 2);
    
    		// TEST for constant arguments
    		checkArgument(argOIs[1] instanceof ConstantObjectInspector);
    		Object constant = ((ConstantObjectInspector) argOIs[1]).getWritableConstantValue();
    		checkArgument(constant instanceof IntWritable);
    		checkArgument(((IntWritable) constant).get() == 1);
    
    		return ObjectInspectorFactory.getStandardStructObjectInspector(
    			Collections.singletonList("col1"),
    			Collections.singletonList(PrimitiveObjectInspectorFactory.javaStringObjectInspector));
    	}
    
    	@Override
    	public void process(Object[] args) throws HiveException {
    		String str = (String) args[0];
    		for (String s : str.split(",")) {
    			forward(s);
    			forward(s);
    		}
    	}
    
    	@Override
    	public void close() {
    	}
    }

从Hive CLI，我们可以看到他们已经注册了：

hive> show functions;
    OK
    ......
    mygenericudf
    myudf
    myudtf

然后使用SQL做查询：

Flink SQL> select mygenericudf(myudf(name), 1) as a, mygenericudf(myudf(age), 1) as b, s from mysourcetable, lateral table(myudtf(name, 1)) as T(s);

**局限性**

*  在Flink中，Hive内置函数目前不支持开箱即用。要使用Hive内置函数，用户必须首先手动在Hive Metastore中注册它们。
 *  基于Blink planner，对Hive功能的支持只对Flink batch进行了测试。
 *  Hive函数目前不能跨Flink中的Catalog使用。
 *  有关数据类型限制，请参考[Flink Table API & SQL - Hive之hive概览][Flink Table API _ SQL - Hive_hive]。

[Flink Table API _ SQL - Catalogs]: https://blog.csdn.net/zpf_940810653842/article/details/100066043
[Flink Table API _ SQL - Hive_hive]: https://blog.csdn.net/zpf_940810653842/article/details/100065614