• modules概念
• 通过hive module使用hive函数
  • 内置函数
  • 自定义函数
• sql 客户端的使用
• 原理分析和源码解析
  • 实现

modules概念

flink 提供了一个module的概念，使用户能扩展flink的内置对象，比如内置函数。这个功能是插件化的，用户可以方便的加载或者移除相应的module。

flink内置了CoreModule，并且提供了一个hive module，允许用户在加载了hive module之后使用hive的函数，包括内置函数、自定义hive函数等等。如果多个module里有重名的函数，则以先加载的函数为准。

用户还可以自定义module，只需要实现Module接口即可。如果是在sql 客户端使用，还需要实现ModuleFactory接口，因为加载的时候，flink会使用SPI机制去匹配获取相应的ModuleFactory，然后实例化相应的moudule。

通过hive module使用hive函数

我们以hive module为例，讲解一下如何使用flink提供的module功能，使用hive module的一些注意事项：

• 通过 Hive Metastore 将带有 UDF 的 HiveCatalog 设置为当前会话的 catalog。
• 将带有 UDF 的 jar 包放入 Flink classpath 中，并在代码中引入。
• 使用 Blink planner，flink 1.11默认就是，不用显示指定

内置函数

• 引入pom

   <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-hive_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>

• 加载module

        String name = "myhive";
        String version = "3.1.2";
        tEnv.loadModule(name, new HiveModule(version));

• 查看module

        System.out.println("list modules ------------------ ");
        String[] modules = tEnv.listModules();
        Arrays.stream(modules).forEach(System.out::println);

运行结果我们看到有两个module

list modules ------------------ 
core
myhive

• 查看函数

    System.out.println("list functions (包含hive函数):------------------  ");
        String[] functions = tEnv.listFunctions();
        Arrays.stream(functions).forEach(System.out::println);

我们看到列出来大概300多个函数，包含flink和hive的内置函数。

• hive函数的使用

在hive里有一个常用的解析json的函数get_json_object，这个可以把json字符串解析之后得到想要的字段，但是flink中没有这个函数，所以我们可以通过这种方式来使用hive的函数，就不用我们自己开发UDF了。

    System.out.println("hive 函数的使用:  ------------------  ");
        String sql = "SELECT data,get_json_object(data, '$.name')  FROM (VALUES ('{\"name\":\"flink\"}'), ('{\"name\":\"hadoop\"}')) AS MyTable(data)";

        List<Row> results = Lists.newArrayList(tEnv.sqlQuery(sql)
                                                   .execute()
                                                   .collect());
        results.stream().forEach(System.out::println);

输出结果：

hive 函数的使用:  ------------------  
{"name":"flink"},flink
{"name":"hadoop"},hadoop

自定义函数

前面我们讲了如何使用hive的内置函数，这个比较简单，接在了hive的module之后就可以用了，还有一种就是如何使用hive的udf函数呢？我们接下来简单聊聊。

• 自定义hive函数

首先我们来自定义一个hive的udf函数

1. 引入pom

    <dependency>
      <groupId>org.apache.hive</groupId>
      <artifactId>hive-exec</artifactId>
      <version>3.1.2</version>
    </dependency>

实现一个自定义函数，就是实现两个int类型数字的加和操作

2. 定义函数

public class TestHiveUDF extends UDF{

    public IntWritable evaluate(IntWritable i,IntWritable j){
        return new IntWritable(i.get() + j.get());
    }

}

完整代码：
https://github.com/zhangjun0x01/bigdata-examples/blob/master/hive/src/main/java/com/test/TestHiveUDF.java

3. 导入

把相应的jar放到hive的classpath下面

定义函数

add jar /home/work/work/hive/lib/hive-1.0-SNAPSHOT.jar; 
CREATE  FUNCTION mysum AS "com.test.TestHiveUDF"; 

4. 测试

        boolean b = Arrays.asList(functions1).contains("mysum");
        System.out.println("是否包含自定义函数： " + b);

        String sqlUdf = "select mysum(1,2)";
        List results1 = Lists.newArrayList(tEnv.sqlQuery(sqlUdf)
                                               .execute()
                                               .collect());
        System.out.println("使用自定义函数处理结果： ");
        results1.stream().forEach(System.out::println);

完整的代码请参考：

https://github.com/zhangjun0x01/bigdata-examples/blob/master/flink/src/main/java/modules/HiveModulesTest.java

sql 客户端的使用

在sql-client-defaults.yaml里配置相关的模块，然后就可以使用了.

# Define modules here.

modules: # note the following modules will be of the order they are specified
  - name: core
    type: core
  - name: hive
    type: hive

原理分析和源码解析

其实相关的源码实现也不难，就是将hive的相关函数转成了flink的函数，我们简单的来看下，主要是在HiveModule类里面。

public class HiveModule implements Module {
    .............
    private final HiveFunctionDefinitionFactory factory;
    private final String hiveVersion;
    private final HiveShim hiveShim;

这个里面有三个主要的变量，用于构造函数的factory，hive的版本hiveVersion，以及用于处理不同版本hive的处理类hiveShim。

实现

具体转换函数的方法是getFunctionDefinition，这个方法调用了工厂类的createFunctionDefinitionFromHiveFunction方法，

我们进入 HiveFunctionDefinitionFactory#createFunctionDefinitionFromHiveFunction。

public FunctionDefinition createFunctionDefinitionFromHiveFunction(String name, String functionClassName) {
        Class clazz;
        try {
            clazz = Thread.currentThread().getContextClassLoader().loadClass(functionClassName);

            LOG.info("Successfully loaded Hive udf '{}' with class '{}'", name, functionClassName);
        } catch (ClassNotFoundException e) {
            throw new TableException(
                String.format("Failed to initiate an instance of class %s.", functionClassName), e);
        }

        if (UDF.class.isAssignableFrom(clazz)) {
            LOG.info("Transforming Hive function '{}' into a HiveSimpleUDF", name);

            return new ScalarFunctionDefinition(
                name,
                new HiveSimpleUDF(new HiveFunctionWrapper<>(functionClassName), hiveShim)
            );
        }
        ..........

我们看到首先会加载相关函数，这个也就是为什么要求我们把hive的udf jar放到flink的classpath的原因。之后是一堆if else判断，Hive UDF 和 GenericUDF 函数会自动转换成 Flink 中的 ScalarFunction，GenericUDTF 会被自动转换成 Flink 中的 TableFunction，UDAF 和 GenericUDAFResolver2 则转换成 Flink 聚合函数(AggregateFunction).这样当我们就可以在flink中使用相应的hive函数了。

参考资料：
[1].https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/hive/hive_functions.html

更多内容，欢迎关注我的公众号【大数据技术与应用实战】