【翻譯】Flink Table Api & SQL — Hive —— Hive 函數

本文翻譯自官網：Hive Functions  https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/table/hive/hive_functions.html

Flink Table Api & SQL 翻譯目錄

用戶可以在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，其中包含該函數作為會話的當前 catalog
• 在Flink的classpath中包含該函數的 jar
• 使用 Blink planner

使用Hive 自定義的函數

假設我們在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中測試了Flink 批處理對Hive功能的支持。

Hive函數當前不能在Flink中的各個 catalog 之間使用。

有關數據類型限制，請參考Hive。 

歡迎關注Flink菜鳥公眾號，會不定期更新Flink（開發技術）相關的推文