By

文章目录
  1. 1. 概述
  2. 2. Java实现UDF
  3. 3. Python实现UDF
  4. 4. Java-UDF vs. Python-UDF
  5. 5. 总结
  6. 6. 参考资料

当业务数据量过大，不能在关系型数据库中统计时，可以通过 Sqoop 导入到 HDFS 上进行统计，类似日志数据一样。导入到 HDFS 上数据，每一条记录通过 \t 或 Ctrl+A 等分割每一个字断，通过 \n 分割每一条记录。然后，通过建立 Hive 表指向 HDFS 对应的路径数据，对 HDFS 上的数据添加 Schema 定义，可以基于 Hive 中 SQL 的语法进行查询统计。

尽管 Hive 提供了丰富的函数，有时仍然不能满足个性化的查询和统计要求。这时，需要用户实现 UDF。

概述

UDF，即 User-Defined Function，用户通过自定义对数据的处理过程，以函数的形式在 HiveQL 中使用。

有 2 种方式在 HiveQL 中实现 UDF。 第一种方式 ，即通过 Java 或其他通过 JVM 实现的语言（Jython、JRuby、Clojure、Groovy），实现UDF； 第二种方式 ，通过 TRANSFORM...AS 语法，使用可执行的脚步语言实现 UDF。 接下来，将分别介绍通过 Java 和 Python 实现 Hive 中的 UDF，最后，比较二者在使用过程中的异同。

假设有一张 Hive 表，描述订单基本信息：

1
2
3
4
5
6
7
8
9
10
11
12
13
CREATE TABLE IF NOT EXISTS order_base(
id                 INT         COMMENT  '自增ID',      
order_sn           STRING      COMMENT  '订单编号',
user_id            INT         COMMENT  '用户ID',
shop_id            INT         COMMENT  '店铺ID',
add_time           INT         COMMENT  '下单时间',
pay_time           INT         COMMENT  '付款时间',
delivery_name      STRING      COMMENT  '收件人姓名',
delivery_address   STRING      COMMENT  '收件人地址',
delivery_phone     STRING      COMMENT  '收件人电话'
)COMMENT '订单基本信息表'
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' LINES TERMINATED BY '\n'
STORED AS TEXTFILE

现在需要查询收件人地址中出现关键字 大学 的记录，找出对应的 (order_sn， delivery_address)，通过 UDF 实现。

Java实现UDF

通过 Java 实现 UDF 时，Java 中的类必须继承类 the org.apache.hadoop.hive.ql.exec.UDF。 并且，UDF 的实现过程必须写在名称为 evaluate 的函数中。由于 evaluate 并未在 UDF 父类中具体说明，因此，用户实现的UDF中，函数 evaluate 的返回类型和参数可以自己指定。当用户实现的 UDF 在 Hive 中执行时，Hive 将会执行函数 evaluate 中的内容。

这里，查询出现指定关键字记录的UDF的Java实现过程如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
package com.whlminds.hive.udf;

import org.apache.hadoop.hive.ql.exec.UDF;

/**
 * @date Oct 7, 2015
 *
 * @author whlminds
 *
 * @Description:  查询出现指定关键字的记录，出现则返回 1，否则返回 0
 *
 */
public class FindSpecifiedWords extends UDF {

	public int evaluate(String targetWords, String rawWords) {
		int found = 0;
		if (rawWords.contains(targetWords)) {
			found = 1;
		}
		return found;
	}
}

由于 org.apache.hadoop.hive.ql.exec.UDF 并非 Java 自带库函数，因此，这里需要引用包含 Hive API的 jar 包，这里是 hive-0.4.1.jar。

用 Java 实现的 UDF 完成了，打包为 hive-udf-java.jar，现在，需要在 HiveQL 中调用。

在 HiveQL 中使用 UDF 前，首先需要添加 jar 包至 Hive 中，对自定义函数 com.whlminds.hive.udf.FindSpecifiedWords 进行注册，可以被 Hive 中脚本使用：

1
hive> ADD JAR /mnt/whlminds/hive-udf-java.jar;

这里，对 com.whlminds.hive.udf.FindSpecifiedWords 起别名，方便调用：

1
CREATE TEMPORARY FUNCTION findwords AS 'com.whlminds.hive.udf.FindSpecifiedWords';

上述步骤，实现了 UDF，在Hive中注册，并起别名方便调用，接下来，就可以在 HiveQL 中使用了：

1
2
3
4
5
SELECT t1.order_sn, t1.delivery_address
FROM order_base t1
WHERE t1.pay_time >= UNIX_TIMESTAMP('2015-10-04 00:00:00')
AND t1.pay_time < UNIX_TIMESTAMP('2015-10-05 00:00:00')
AND findwords('大学', t1.delivery_address) = 1;

为查询方便，可以将 Hive 中注册，起别名，以及调用过程一起写在SQL脚本 find_specified_order.sql 中：

1
2
3
4
5
6
7
8
9
ADD JAR /mnt/whlminds/hive-udf-java.jar;

CREATE TEMPORARY FUNCTION findwords AS 'com.whlminds.hive.udf.FindSpecifiedWords';

SELECT t1.order_sn, t1.delivery_address
FROM order_base t1
WHERE t1.pay_time >= UNIX_TIMESTAMP('2015-10-04 00:00:00')
AND t1.pay_time < UNIX_TIMESTAMP('2015-10-05 00:00:00')
AND findwords('大学', t1.delivery_address) = 1;

然后，执行 SQL 脚本：

1
hive> hive -f find_specified_order.sql

Python实现UDF

通过 Python 实现 Hive 的 UDF，Python 脚本需要以特定的方式读入和输出，除了必须引用 sys 包外，无须引用其他外部包。

这里，查询出现指定关键字记录的 UDF 的 Python 实现过程 FindSpecifiedWords.py 如下：

1
2
3
4
5
6
7
8
9
10
11
12
#!/usr/bin/python

import sys

for line in sys.stdin:
    order_sn, delivery_address = line.strip().split("\t")

    found = "N"
    pos = delivery_address.decode("utf8").find(u'\u5927\u5b66')
    if(pos > -1):
        found = "Y"
    print "\t".join([order_sn, delivery_address, found])

这里，Python 实现的 UDF，需要批量的读入数据，并一对一的批量输出。其中，u'\u5927\u5b66' 是 大学 的 utf8 的编码。

使用 Python 实现的 UDF 完成后，需要通过 ADD FILE 指令添加至 Hive 中进行注册，无需起别名：

1
hive > ADD FILE /mnt/whlminds/FindSpecifiedWords.py

注册完后，Python 实现的 UDF 就可以通过 TRANSFORM...AS 在 HiveQL 中使用，语法如下：

1
2
3
4
SELECT TRANSFORM (<columns>)
USING 'python <python_script>'
AS (<columns>)
FROM <table>;

其中， SELECT 中的 columns 是 FROM 中 table 的列名， 而 AS 中的 columns 是经过 USING 中 Python 脚本 python_script 计算返回的列名。

这里，查找包含指定关键字的 HiveQL 脚本如下：

1
2
3
4
5
6
7
8
9
10
11
SELECT t2.order_sn, t2.delivery_address
FROM

(SELECT TRANSFORM (t1.order_sn, t1.delivery_address)
USING 'python FindSpecifiedWords.py'
AS (order_sn STRING, delivery_address STRING, found STRING)
FROM order_base t1
WHERE t1.pay_time >= UNIX_TIMESTAMP('2015-10-04 00:00:00')
AND t1.pay_time < UNIX_TIMESTAMP('2015-10-05 00:00:00')) t2

WHERE t2.found = 'Y';

为查询方便，可以将 Hive 中注册，以及调用过程一起写在 SQL 脚本 find_specified_order.sql 中：

1
2
3
4
5
6
7
8
9
10
11
12
13
ADD FILE /mnt/whlminds/FindSpecifiedWords.py;

SELECT t2.order_sn, t2.delivery_address
FROM

(SELECT TRANSFORM (t1.order_sn, t1.delivery_address)
USING 'python FindSpecifiedWords.py'
AS (order_sn STRING, delivery_address STRING, found STRING)
FROM order_base t1
WHERE t1.pay_time >= UNIX_TIMESTAMP('2015-10-04 00:00:00')
AND t1.pay_time < UNIX_TIMESTAMP('2015-10-05 00:00:00')) t2

WHERE t2.found = 'Y';

然后，执行 SQL 脚本：

1
hive> hive -f find_specified_order.sql

Java-UDF vs. Python-UDF

上述2部分分别介绍了通过 Java 和 Python 实现 Hive 中 UDF 过程，这里比较下二者的异同：

  • Java 实现 UDF，需要引用包含 Hive API 的外部 jar 包，而 Python 无需引起其他外部包；
  • Java 实现 UDF 后，需要打包后才可被 HiveQL 调用，而通过 Python 实现 UDF 后，可以在 HiveQL 中直接被调用；
  • Java 实现 UDF，对读入和输出数据方式没有要求，实现的 UDF 可以输入一条记录的指定列数据，输出结果可以直接在 HiveQL 的 WHERE 中用于判断条件使用；
  • Python 实现的 UDF，对读入和输出数据方式有特殊要求，需要对 HiveQL 中表的指定列数据批量读入，然后一对一地批量输出，因此，通过 Python 实现的 UDF 可以结合子查询使用。

总结

Hive 的 UDF 是对 Hive 内置的函数的扩展，本文主要介绍通过 Java 和 Python 实现 Hive 中的 UDF 过程，最后，比较了二者在使用过程中异同。

参考资料

  1. Hive Operators and User-Defined Functions (UDFs)
  2. A Guide To User-Defined Functions In Apache Hive
  3. Use Python with Hive and Pig in HDInsight
文章目录
  1. 1. 概述
  2. 2. Java实现UDF
  3. 3. Python实现UDF
  4. 4. Java-UDF vs. Python-UDF
  5. 5. 总结
  6. 6. 参考资料