devbox
笔触狂放9
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

44、Flink之module模块介绍及使用示例和Flink SQL使用hive内置函数及自定义函数详细示例--网上有些说法好像是错误的

作者：笔触狂放9 | 2024-05-29 09:47:05

赞

踩

flink之module模块介绍及使用示例和flink sql使用hive内置函数及自定义函数详细示

Flink 系列文章

一、Flink 专栏

Flink 专栏系统介绍某一知识点，并辅以具体的示例进行说明。

  • 1、Flink 部署系列
    本部分介绍Flink的部署、配置相关基础内容。

  • 2、Flink基础系列
    本部分介绍Flink 的基础部分，比如术语、架构、编程模型、编程指南、基本的datastream api用法、四大基石等内容。

  • 3、Flik Table API和SQL基础系列
    本部分介绍Flink Table Api和SQL的基本用法，比如Table API和SQL创建库、表用法、查询、窗口函数、catalog等等内容。

  • 4、Flik Table API和SQL提高与应用系列
    本部分是table api 和sql的应用部分，和实际的生产应用联系更为密切，以及有一定开发难度的内容。

  • 5、Flink 监控系列
    本部分和实际的运维、监控工作相关。

二、Flink 示例专栏

Flink 示例专栏是 Flink 专栏的辅助说明，一般不会介绍知识点的信息，更多的是提供一个一个可以具体使用的示例。本专栏不再分目录，通过链接即可看出介绍的内容。

两专栏的所有文章入口点击：Flink 系列文章汇总索引

文章目录

本文介绍了Flink的module功能以及Flink SQl使用hive的内置函数和hive的自定义函数功能。
本文依赖hadoop、hive、flink集群能正常使用，其版本分别是3.1.4、3.1.2和1.13.6，内容是按照1.17版本写的。
本文分为2个部分，即介绍了Flink 的Module和Flink SQL 使用Hive的内置函数及自定义函数，并提供了完整的可验证通过的示例。

一、模块Modules

1、模块介绍

模块允许用户扩展 Flink 的内置对象，例如定义行为类似于 Flink 内置函数的函数。它们是可插拔的，虽然 Flink 提供了一些预构建的模块，但用户可以编写自己的模块。

例如，用户可以定义自己的地理函数，并将它们作为内置函数插入 Flink 中，用于 Flink SQL 和表 API。另一个例子是用户可以加载一个现成的 Hive 模块，将 Hive 内置函数用作 Flink 内置函数。

此外，模块可以提供内置的表源（ table source）和接收器工厂（sink factories），这些工厂禁用了基于 Java 服务提供程序接口 （SPI） 的 Flink 默认发现机制，或者影响如何在没有相应目录的情况下创建临时表的连接器。

模块提供的对象被视为 Flink 系统（内置）对象的一部分;因此，它们没有任何命名空间。

2、模块类别Module Types

1）、CoreModule

CoreModule 包含 Flink 的所有系统（内置）功能，默认情况下加载并启用。

2）、HiveModule

HiveModule 为 SQL 和 Table API 用户提供了 Hive 内置函数作为 Flink 的系统函数。Flink 的 Hive 文档提供了有关设置模块的完整详细信息。

3）、User-Defined Module

用户可以通过实现模块接口来开发自定义模块。若要在 SQL CLI 中使用自定义模块，用户应通过实现模块工厂接口来开发模块及其相应的模块工厂。

模块工厂定义一组属性，用于在 SQL CLI 引导时配置模块。属性将传递给发现服务，该服务尝试将属性与 ModuleFactory 匹配并实例化相应的模块实例。

3、模块生命周期和解析顺序Module Lifecycle and Resolution Order

可以加载、启用、禁用和卸载模块。当 TableEnvironment 最初加载模块时，默认情况下会启用该模块。Flink 支持多个模块，并跟踪加载顺序以解析元数据。此外，Flink 只解析启用模块之间的功能。例如，当两个模块中存在两个同名的函数时，将有三个条件，如下。

  • 如果两个模块都启用了，那么 Flink 会根据模块的解析顺序解析函数。
  • 如果其中一个被禁用，则 Flink 会将函数解析为启用的模块。
  • 如果两个模块都被禁用，那么 Flink 就无法解析该功能。

用户可以通过使用不同声明顺序的模块来更改解析顺序。例如，用户可以指定 Flink 通过 USE MODULES Hive、core 首先在 Hive 中查找函数。

此外，用户还可以通过不声明模块来禁用模块。例如，用户可以指定 Flink 通过 USE MODULES hive 禁用核心模块（但是，强烈建议不要禁用核心模块）。禁用模块不会卸载它，用户可以使用它时再次启用它。例如，用户可以带回核心模块并将其放置在第一个通过USE MODULES core，hive。仅当模块已加载时，才能启用该模块。使用卸载的模块将引发异常。最终，用户可以卸载模块。

禁用和卸载模块的区别在于，TableEnvironment 仍然保留已禁用的模块，用户可以列出所有已加载的模块以查看已禁用的模块。

4、模块Modules的使用

一般有2种使用方式，即sql方式和开发语言编程方式，如java、scala、python。
用户可以使用 SQL 在表 API 和 SQL CLI 中加载/卸载/使用/列出模块。
用户可以使用 Java、Scala 或 Python 以编程方式加载/卸载/使用/列出模块。

1）、SQL方式

sql的方式本文列出了三种，即通过客户端直接使用、java语言中编写sql和配置文件方式。

  • SQL cli
Flink SQL> SHOW MODULES;
+-------------+
| module name |
+-------------+
|        core |
+-------------+
1 row in set

Flink SQL> SHOW FULL MODULES;
+-------------+------+
| module name | used |
+-------------+------+
|        core | true |
+-------------+------+
1 row in set

Flink SQL> LOAD MODULE hive WITH ('hive-version' = '3.1.2');
[INFO] Execute statement succeed.

Flink SQL> SHOW MODULES;
+-------------+
| module name |
+-------------+
|        core |
|        hive |
+-------------+
2 rows in set

Flink SQL> SHOW FULL MODULES;
+-------------+------+
| module name | used |
+-------------+------+
|        core | true |
|        hive | true |
+-------------+------+
2 rows in set

Flink SQL> USE MODULES hive, core ;
[INFO] Execute statement succeed.

Flink SQL> SHOW MODULES;
+-------------+
| module name |
+-------------+
|        hive |
|        core |
+-------------+
2 rows in set

Flink SQL> SHOW FULL MODULES;
+-------------+------+
| module name | used |
+-------------+------+
|        hive | true |
|        core | true |
+-------------+------+
2 rows in set

Flink SQL> UNLOAD MODULE hive;
[INFO] Execute statement succeed.

Flink SQL> SHOW MODULES;
+-------------+
| module name |
+-------------+
|        core |
+-------------+
1 row in set

Flink SQL> SHOW FULL MODULES;
+-------------+------+
| module name | used |
+-------------+------+
|        core | true |
+-------------+------+
1 row in set


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • java
EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
TableEnvironment tableEnv = TableEnvironment.create(settings);

// Show initially loaded and enabled modules
tableEnv.executeSql("SHOW MODULES").print();
// +-------------+
// | module name |
// +-------------+
// |        core |
// +-------------+
tableEnv.executeSql("SHOW FULL MODULES").print();
// +-------------+------+
// | module name | used |
// +-------------+------+
// |        core | true |
// +-------------+------+

// Load a hive module
tableEnv.executeSql("LOAD MODULE hive WITH ('hive-version' = '...')");

// Show all enabled modules
tableEnv.executeSql("SHOW MODULES").print();
// +-------------+
// | module name |
// +-------------+
// |        core |
// |        hive |
// +-------------+

// Show all loaded modules with both name and use status
tableEnv.executeSql("SHOW FULL MODULES").print();
// +-------------+------+
// | module name | used |
// +-------------+------+
// |        core | true |
// |        hive | true |
// +-------------+------+

// Change resolution order
tableEnv.executeSql("USE MODULES hive, core");
tableEnv.executeSql("SHOW MODULES").print();
// +-------------+
// | module name |
// +-------------+
// |        hive |
// |        core |
// +-------------+
tableEnv.executeSql("SHOW FULL MODULES").print();
// +-------------+------+
// | module name | used |
// +-------------+------+
// |        hive | true |
// |        core | true |
// +-------------+------+

// Disable core module
tableEnv.executeSql("USE MODULES hive");
tableEnv.executeSql("SHOW MODULES").print();
// +-------------+
// | module name |
// +-------------+
// |        hive |
// +-------------+
tableEnv.executeSql("SHOW FULL MODULES").print();
// +-------------+-------+
// | module name |  used |
// +-------------+-------+
// |        hive |  true |
// |        core | false |
// +-------------+-------+

// Unload hive module
tableEnv.executeSql("UNLOAD MODULE hive");
tableEnv.executeSql("SHOW MODULES").print();
// Empty set
tableEnv.executeSql("SHOW FULL MODULES").print();
// +-------------+-------+
// | module name |  used |
// +-------------+-------+
// |        hive | false |
// +-------------+-------+

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • yaml
    使用 YAML 定义的所有模块都必须提供指定类型的类型属性。现支持以下类型
    在这里插入图片描述
modules:
   - name: core
     type: core
   - name: hive
     type: hive

  • 1
  • 2
  • 3
  • 4
  • 5

使用SQL方式时，模块的名称是用于加载模块的，所以是模块的唯一标识，并且大小写敏感

2）、编码方式-java

EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
TableEnvironment tableEnv = TableEnvironment.create(settings);

// Show initially loaded and enabled modules
tableEnv.listModules();
// +-------------+
// | module name |
// +-------------+
// |        core |
// +-------------+
tableEnv.listFullModules();
// +-------------+------+
// | module name | used |
// +-------------+------+
// |        core | true |
// +-------------+------+

// Load a hive module
tableEnv.loadModule("hive", new HiveModule());

// Show all enabled modules
tableEnv.listModules();
// +-------------+
// | module name |
// +-------------+
// |        core |
// |        hive |
// +-------------+

// Show all loaded modules with both name and use status
tableEnv.listFullModules();
// +-------------+------+
// | module name | used |
// +-------------+------+
// |        core | true |
// |        hive | true |
// +-------------+------+

// Change resolution order
tableEnv.useModules("hive", "core");
tableEnv.listModules();
// +-------------+
// | module name |
// +-------------+
// |        hive |
// |        core |
// +-------------+
tableEnv.listFullModules();
// +-------------+------+
// | module name | used |
// +-------------+------+
// |        hive | true |
// |        core | true |
// +-------------+------+

// Disable core module
tableEnv.useModules("hive");
tableEnv.listModules();
// +-------------+
// | module name |
// +-------------+
// |        hive |
// +-------------+
tableEnv.listFullModules();
// +-------------+-------+
// | module name |  used |
// +-------------+-------+
// |        hive |  true |
// |        core | false |
// +-------------+-------+

// Unload hive module
tableEnv.unloadModule("hive");
tableEnv.listModules();
// Empty set
tableEnv.listFullModules();
// +-------------+-------+
// | module name |  used |
// +-------------+-------+
// |        hive | false |
// +-------------+-------+

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81

二、Hive Functions内置函数和自定义函数使用

关于Flink sql使用hive的内置函数可以参考文章：41、Flink之Hive 方言介绍及详细示例

1、通过 HiveModule 使用 Hive 内置函数

HiveModule 为 Flink SQL 和 Table API 用户提供了 Hive 内置函数作为 Flink 系统（内置）函数。

  • java 代码示例
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

		String moduleName            = "myhive";
		String hiveVersion         = "3.1.2";
		tenv.loadModule(moduleName, new HiveModule(hiveVersion));

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • yaml配置示例
modules:
   - name: core
     type: core
   - name: myhive
     type: hive

  • 1
  • 2
  • 3
  • 4
  • 5
  • hive内置函数使用示例
CREATE CATALOG alan_hivecatalog WITH (
    'type' = 'hive',
    'default-database' = 'testhive',
    'hive-conf-dir' = '/usr/local/bigdata/apache-hive-3.1.2-bin/conf'
);
use catalog alan_hivecatalog;
set table.sql-dialect=hive;
load module hive;
use modules hive,core;
select explode(array(1,2,3));
create table tbl (key int,value string);
set execution.runtime-mode=streaming; 
insert into table tbl values (5,'e'),(1,'a'),(1,'a'),(3,'c'),(2,'b'),(3,'c'),(3,'c'),(4,'d');
select * from tbl;

--------------------flink sql 操作
Flink SQL> select explode(array(1,2,3));
Hive Session ID = 7d3ae2d5-24f3-4d97-9897-83c8a9abda9b
[ERROR] Could not execute SQL statement. Reason:
org.apache.hadoop.hive.ql.parse.SemanticException: Invalid function explode

Flink SQL> set table.sql-dialect=hive;

Flink SQL> select explode(array(1,2,3));
Hive Session ID = c0b87333-4957-4c18-b197-27649a3f2ae2
[ERROR] Could not execute SQL statement. Reason:
org.apache.hadoop.hive.ql.parse.SemanticException: Invalid function explode

Flink SQL> load module hive;

Flink SQL> use modules hive,core;

Flink SQL> select explode(array(1,2,3));

+----+-------------+
| op |         col |
+----+-------------+
| +I |           1 |
| +I |           2 |
| +I |           3 |
+----+-------------+
Received a total of 3 rows

Flink SQL> create table tbl (key int,value string);

Flink SQL> insert overwrite table tbl values (5,'e'),(1,'a'),(1,'a'),(3,'c'),(2,'b'),(3,'c'),(3,'c'),(4,'d');
Hive Session ID = 12fe08fa-5e63-44b2-8fc3-a90064959451
[INFO] Submitting SQL update statement to the cluster...
[ERROR] Could not execute SQL statement. Reason:
java.lang.IllegalStateException: Streaming mode not support overwrite.

Flink SQL> set execution.runtime-mode=batch; 
Hive Session ID = 4f17cc70-165c-4540-a299-874b66458521
[INFO] Session property has been set.

Flink SQL> insert overwrite table tbl values (5,'e'),(1,'a'),(1,'a'),(3,'c'),(2,'b'),(3,'c'),(3,'c'),(4,'d');
Hive Session ID = 1923623f-03d3-44b4-93ab-ee8498c5da06
[INFO] Submitting SQL update statement to the cluster...
[ERROR] Could not execute SQL statement. Reason:
java.lang.IllegalArgumentException: Checkpoint is not supported for batch jobs.

Flink SQL> set execution.runtime-mode=streaming; 

Flink SQL> insert into table tbl values (5,'e'),(1,'a'),(1,'a'),(3,'c'),(2,'b'),(3,'c'),(3,'c'),(4,'d');

Flink SQL> select * from tbl;
+----+-------------+--------------------------------+
| op |         key |                          value |
+----+-------------+--------------------------------+
| +I |           5 |                              e |
| +I |           1 |                              a |
| +I |           1 |                              a |
| +I |           3 |                              c |
| +I |           2 |                              b |
| +I |           3 |                              c |
| +I |           3 |                              c |
| +I |           4 |                              d |
+----+-------------+--------------------------------+
Received a total of 8 rows


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80

一些hive的内置函数存在线程安全问题，建议更新hive的版本修复它

2、使用原生的hive 聚合函数Native Hive Aggregate Functions

如果 HiveModule 加载的优先级高于 CoreModule，Flink 会先尝试使用 Hive 内置函数。然后对于 Hive 内置的聚合函数，Flink 现在只能使用基于排序的聚合算子。从 Flink 1.17 开始，我们引入了一些原生的 Hive 聚合函数，可以使用基于哈希的聚合运算符来执行。目前仅支持5个函数，即sum/count/avg/min/max，未来将支持更多聚合函数。用户可以通过打开选项 table.exec.hive.native-agg-function.enabled 来使用原生的聚合函数，这为作业带来了显著的性能改进。
在这里插入图片描述

原生的聚合函数（native aggregation functions）的功能现在与 Hive 内置聚合函数不完全一致，例如，不支持某些数据类型。如果性能不是瓶颈，则无需启用此选项。此外，通过 SqlClient 使用时，无法为每个作业打开 table.exec.hive.native-agg-function.enabled 选项，目前仅支持模块级别。用户应先启用此选项，然后加载 HiveModule。此问题将在将来修复。

3、hive的自定义函数介绍

用户可以在 Flink 中使用他们现有的 Hive 用户定义函数。
当前支持的用户自定义函数包括如下：

  • UDF
  • GenericUDF
  • GenericUDTF
  • UDAF
  • GenericUDAFResolver2

在查询计划和执行时，Hive 的 UDF 和 GenericUDF 会自动转换为 Flink 的 ScalarFunction，Hive 的 GenericUDTF 会自动转换为 Flink 的 TableFunction，Hive 的 UDAF 和 GenericUDAFResolver2 会自动转换为 Flink 的 AggregateFunction。

若要使用 Hive 用户定义函数，前提条件如下：

  • 设置由 Hive metastore支持的 HiveCatalog，该目录包含该函数作为会话的当前目录
  • 在 Flink 的类路径中包含包含该函数的 jar
  • 使用 Blink planner （1.14版本以后没有这个限制）

4、hive的自定义函数使用-示例

关于hive自定义函数的开发过程详见文章关于自定义函数的部分：7、hive shell客户端与属性配置、内置运算符、函数（内置运算符与自定义UDF运算符）
简单来说分为如下几步：

  • 写一个java类，继承UDF，并重载evaluate方法，方法中实现函数的业务逻辑（重载意味着可以在一个java类中实现多个函数功能）
  • 程序打成jar包，上传HS2服务器本地或者HDFS
  • 客户端命令行中添加jar包到Hive的classpath： hive>add JAR /xxxx/udf.jar
  • 注册成为临时函数（给UDF命名）：create temporary function 函数名 as ‘UDF类全路径’
  • HQL中使用函数

1）、定义函数

本函数实现功能如下：

  • 能够对输入数据进行非空判断、手机号位数判断
  • 能够实现校验手机号格式，把满足规则的进行****处理
  • 对于不符合手机号规则的数据直接返回，不处理
  • 代码 -有两种实现方式即UDF和GenericUDF
1、UDF实现
public class EncryptPhoneNumber extends UDF {
    /**
     * 重载evaluate方法 实现函数的业务逻辑
     * @param phoNum  入参：未加密手机号
     * @return 返回：加密后的手机号字符串
     */
    public String evaluate(String phoNum){
        String encryptPhoNum = null;
        //手机号不为空 并且为11位
        if (StringUtils.isNotEmpty(phoNum) && phoNum.trim().length() == 11 ) {
            //判断数据是否满足中国大陆手机号码规范
                String regex = "^(1[3-9]\\d{9}$)";
            Pattern p = Pattern.compile(regex);
            Matcher m = p.matcher(phoNum);
            if (m.matches()) {//进入这里都是符合手机号规则的
                //使用正则替换 返回加密后数据
                encryptPhoNum = phoNum.trim().replaceAll("(\\d{3})\\d{4}(\\d{4})","$1****$2");
            }else{
                //不符合手机号规则 数据直接原封不动返回
                encryptPhoNum = phoNum;
            }
        }else{
            //不符合11位 数据直接原封不动返回
            encryptPhoNum = phoNum;
        }
        return encryptPhoNum;
    }
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
2、GenericUDF实现
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaStringObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.StringObjectInspector;


public class EncryptPhoneNumber extends GenericUDF {
	StringObjectInspector elementOI;
	/**
	 * Initialize this GenericUDF. This will be called once and only once per
	 * GenericUDF instance.
	 */
	@Override
	public ObjectInspector initialize(ObjectInspector[] arguments) throws UDFArgumentException {
		// 1. 检查该记录是否传过来正确的参数数量
		if (arguments.length != 1) {
			throw new UDFArgumentException("输入参数错误，必须是一个参数。");
		}
		// 2. 检查该条记录是否传过来正确的参数类型
		ObjectInspector a = arguments[0];
		if (!(a instanceof StringObjectInspector)) {
			throw new UDFArgumentException("輸入參數錯誤，需要是一個字符串");
		}

		// 3. 检查通过后，将参数赋值给成员变量ObjectInspector，为了在evaluate()中使用
		this.elementOI = (StringObjectInspector) a;
		return PrimitiveObjectInspectorFactory.javaStringObjectInspector;
	}

	/**
	 * Evaluate the GenericUDF with the arguments. 重载evaluate方法 实现函数的业务逻辑
	 */
	@Override
	public Object evaluate(DeferredObject[] arguments) throws HiveException {
		String phoNum = elementOI.getPrimitiveJavaObject(arguments[0].get()).toString();
		String encryptPhoNum = null;
		// 手机号不为空 并且为11位
		if (StringUtils.isNotEmpty(phoNum) && phoNum.trim().length() == 11) {
			// 判断数据是否满足中国大陆手机号码规范
			String regex = "^(1[3-9]\\d{9}$)";
			Pattern p = Pattern.compile(regex);
			Matcher m = p.matcher(phoNum);
			if (m.matches()) {// 进入这里都是符合手机号规则的
				// 使用正则替换 返回加密后数据
				encryptPhoNum = phoNum.trim().replaceAll("(\\d{3})\\d{4}(\\d{4})", "$1****$2");
			} else {
				// 不符合手机号规则 数据直接原封不动返回
				encryptPhoNum = phoNum;
			}
		} else {
			// 不符合11位 数据直接原封不动返回
			encryptPhoNum = phoNum;
		}
		return encryptPhoNum;
	}

	/**
	 * Get the String to be displayed in explain.
	 */
	@Override
	public String getDisplayString(String[] children) {
		return "this is a EncryptPhoneNumber pro.";
	}

	public static void main(String[] args) throws Exception {
		EncryptPhoneNumber ep = new EncryptPhoneNumber();
		JavaStringObjectInspector stringOI = PrimitiveObjectInspectorFactory.javaStringObjectInspector;

		JavaStringObjectInspector resultInspector = (JavaStringObjectInspector) ep.initialize(new ObjectInspector[] { stringOI });

		Object result = ep.evaluate(new DeferredObject[] { new DeferredJavaObject("13917885967") });
		System.out.println("result:" + result);
	}

}


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82

2）、hive中注册函数并使用

本处简单的描述过程和命令。

  • 打包
mvn package -Dmaven.test.skip=true

  • 1
  • 添加jar包到Hive的classpath
0: jdbc:hive2://server4:10000> add jar /usr/local/bigdata/testjar/hive-0.0.1-SNAPSHOT.jar;
No rows affected (0.01 seconds)


  • 1
  • 2
  • 3
  • 注册成为永久函数

该处需要特别注意，同时需要注册函数的时候带上数据库名称，否则默认为default.函数名称，如：default.encryptphonenumber

-- alan_testdatabase 为hive中一个数据库名称
CREATE FUNCTION alan_testdatabase.encryptPhoneNumber AS 'org.hive.udf.EncryptPhoneNumber';

0: jdbc:hive2://server4:10000> CREATE FUNCTION alan_testdatabase.encryptPhoneNumber AS 'org.hive.udf.EncryptPhoneNumber';
No rows affected (0.023 seconds)


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 验证hive的自定义函数功能
0: jdbc:hive2://server4:10000> select alan_testdatabase.encryptPhoneNumber("13788889999");
+--------------+
|     _c0      |
+--------------+
| 137****9999  |
+--------------+


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7

3）、flink sql中使用

前提：需要将hive 自定义函数的jar包（也就是第二步中注册为函数的那个jar包）放在flink的lib目录下，并重启集群。

  • 在hive中查看自定义的函数
0: jdbc:hive2://server4:10000> show functions;
+---------------------------------------+
|               tab_name                |
+---------------------------------------+
......
| aes_decrypt                           |
| aes_encrypt                           |
| alan_testdatabase.encryptphonenumber  |
| and                                   |
| array                                 |
........

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 设置Flink sql的环境

Flink SQL> LOAD MODULE hive WITH ('hive-version' = '3.1.2');
[INFO] Execute statement succeed.

Flink SQL> use modules hive,core;
[INFO] Execute statement succeed.

Flink SQL> SHOW FULL MODULES;
+-------------+------+
| module name | used |
+-------------+------+
|        hive | true |
|        core | true |
+-------------+------+
2 rows in set

Flink SQL> use catalog alan_hivecatalog;
[INFO] Execute statement succeed.

Flink SQL> set table.sql-dialect=hive;
[INFO] Session property has been set.

Flink SQL> SET sql-client.execution.result-mode = tableau;
[INFO] Session property has been set.


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 在flink中查看函数

Flink SQL> use alan_testdatabase;
[INFO] Execute statement succeed.

Flink SQL> show functions;
Hive Session ID = 5d34cbf8-5984-4ec0-8527-e06a948ad7ca
+--------------------------------+
|                  function name |
+--------------------------------+
.........
|             encryptphonenumber |
.........

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 在flink sql中验证hive 的自定义函数
Flink SQL> select alan_testdatabase.encryptPhoneNumber("13788889999");
+----+--------------------------------+
| op |                         _o__c0 |
+----+--------------------------------+
| +I |                    137****9999 |
+----+--------------------------------+
Received a total of 1 row

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7

以上，介绍了Flink的module功能以及Flink SQl使用hive的内置函数和hive的自定义函数功能。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/笔触狂放9/article/detail/641596
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号