Hive 的 UDF 函数（Python 版本）

Hive 的 UDF 函数（Python 版本）

一、UDF 是什么

Hive 中支持很多的函数，可以极大的方便我们的工作。但是系统内置的函数虽然很多，却不能满足我们所有的需求。在遇到一些个性化需求的时候，系统内置的函数无法满足，此时就需要使用代码来完成自己的业务逻辑。

UDF: User Defined Function，字面来理解就是用户自定义的函数。其实展开来说，很多系统内置的函数也属于是 UDF 函数。UDF 函数的最大特点是“一对一”。即：一行输入、一行输出。例如，

month

函数可以获取到一个日期中的月份。如果我们使用

month(birthday)

就可以获取到一个生日字段中的月份。一行数据中的生日的数据，带入到

month

函数中，就可以得到一行的月份信息。

UDAF: User Defined Aggregation Function，字面来理解就是用户自定义的聚合函数。其实展开来说，常见的聚合函数也属数据 UDAF 函数的。例如

sum

,

count

,

max

,

min

,

avg

，这些函数是作用于某一列的多行数据的。接受多行数据的输入，输出一行的聚合结果。

UDTF: User Defined Table Generate Function，字面来理解就是用户自定义的表生成函数。这种函数可以接受一行的输入数据，输出多行的结果。例如 Hive 中的

explode

就是这样的函数，可以接受一个数组或者字典的数据，将其中的数据展开成多行。

二、UDF 函数的实现

Hive 本身是用 Java 语言来编写的，提供的 UDF 函数的接口也是 Java 的编程接口。但是 Hive 也同时提供了对 Python 的支持，使得可以使用 Python 的语言来编写自己的 UDF、UDAF、UDTF 的逻辑，并且实现起来比 Java 的更加简单！

本文只是记录 UDF 函数的 Python 实现版本！

Python 实现的 UDF、UDAF、UDTF 函数，需要对输入的一整行的数据进行处理，而并不是针对某一个字段进行处理！

• 读取每一行的输入：可以使用 Python 中的标准输入流来读取。需要注意的是，无论 Hive 的表中是以什么样的字符进行分隔的，这里读取到的字段之间始终是以 \t 进行分隔的！因此我们也必须要使用 \t 进行字段的切割。
• 输出一行的数据：在将一整行的数据处理完成之后，如果需要输出的时候，直接使用 print 函数打印即可。与输入的一样，输出的不同的字段之间也必须要使用 \t 进行分隔！
• UDAF 函数实现、UDTF 函数实现：如果实现 UDAF、UDTF 这种输入行和输出行不一致的情况，直接控制 print 函数的输出次数即可。一次的 print 就会生成一行的数据，因此合理控制 print 函数的调用次数，即可实现 UDAF、UDTF 函数。

三、UDF 函数案例

案例一、将字母转成全部大写

1、准备数据文件

zhangsan,male,19
lisi,female,20
wangwu,male,18
zhaoliu,female,19

2、创建数据表

createtableifnotexists example_01(
    t_name string,
    t_gender string,
    t_age int)row format delimited
fieldsterminatedby',';

3、代码编写

# -*- coding: utf-8 -*-# @Author   : 大数据章鱼哥# @Company  : 北京千锋互联科技有限公司# string_upper.pyimport sys

for line in sys.stdin:print(line.upper(), end='')

4、应用函数

-- 应用UDFaddfile string_upper.py

-- 查询名字，将其转成大写select transform(t_name)using'python3 string_upper.py'as(t_name)from example_01;-- 查询名字、性别，将其转成大写select transform(t_name, t_gender)using'python3 string_upper.py'as(t_name, t_gender)from example_01;

案例二、只需要将姓名转成大写

1、数据准备

数据文件和表，沿用案例一即可

2、代码编写

# -*- coding: utf-8 -*-# @Author   : 大数据章鱼哥# @Company  : 北京千锋互联科技有限公司# string_name_upper.pyimport sys

for line in sys.stdin:# 1. 切割出来每一列
    columns = line.strip().split('\t')# 2. 判断切割出来的列的数量，如果数量不正确，这一行数据不处理
    length =len(columns)
    name = columns[0].upper()if length >=1else'NULL'
    gender = columns[1]if length >=2else'NULL'
    age = columns[2]if length >=3else'NULL'# 3. 打印处理之后的结果print('\t'.join([name, gender, age]))

3、应用函数

-- 查询所有字段，将名字转成大写select transform(*)using'python3 string_name_upper.py'as(name, gender, age)from example_01;+-----------+---------+------+|   name    | gender  | age  |+-----------+---------+------+| ZHANGSAN  | male    |19|| LISI      | female  |20|| WANGWU    | male    |18|| ZHAOLIU   | female  |19|+-----------+---------+------+

案例三、身份证解析

1、准备数据文件

身份证号码是按照身份证的规则随机生成，如有雷同，纯属巧合。

张三三,410023198911223344
李思思,110011200210103232
汪呜呜,37126520000223987X
赵溜溜,561276199512019866

2、准备数据表

createtableifnotexists example_02(
    username string,
    idcard string
)row format delimited
fieldsterminatedby',';

3、 代码编写

需求：通过表中的数据，查询出 姓名、身份证号、出生年月日、年龄、性别

# -*- coding: utf-8 -*-# @Author   : 大数据章鱼哥# @Company  : 北京千锋互联科技有限公司# 身份证验证#   如果身份证不是一个合法的身份证，输出空#   如果身份证是合法的，解析出年龄、生日、性别import sys
import re
import datetime

defcalculate_age(year, month, day):
    now = datetime.datetime.now()
    age = now.year - year
    if now.month < month:
        age -=1elif now.month == month and now.day < day:
        age -=1return age

for line in sys.stdin:# 1. 切割出来每一个组成部分
    fields = line.strip().split('\t')# 2. 如果长度不到两位，说明数据有问题，不做任何处理iflen(fields)!=2:continue# 3. 提取姓名和身份证
    name = fields[0]
    id_card = fields[1]# 4. 身份证正则匹配
    m = re.fullmatch(r'(\d{6})(?P<year>(19|20)\d{2})(?P<month>0[1-9]|1[0-2])(?P<day>[012][0-9]|10|20|30|31)\d{2}(?P<gender>\d)[0-9xX]', id_card)if m isNone:print('\t'.join([name, id_card,'NULL','NULL','NULL']))else:# 出生年月日
        year = m.group('year')
        month = m.group('month')
        day = m.group('day')
        age = calculate_age(int(year),int(month),int(day))# 计算性别
        gender ='男'ifint(m.group('gender'))%2!=0else'女'# 拼接生日
        birthday =f'{year}-{month}-{day}'print('\t'.join([name, id_card, birthday,str(age), gender]))

4、应用函数

-- 添加文件addfile idcard_parser.py
-- 执行查询select transform(*)using'python3 idcard_parser.py'as(name, idcard, birthday, age, gender)from example_02
    
+-------+---------------------+-------------+------+---------+| name  |       idcard        |  birthday   | age  | gender  |+-------+---------------------+-------------+------+---------+| 张三三 |410023198911223344|1989-11-22|33| 女       || 李思思 |110011200210103232|2002-10-10|20| 男       || 汪呜呜 |37126520000223987X  |2000-02-23|22| 男       || 赵溜溜 |561276199512019866|1995-12-01|27| 女       |+-------+---------------------+-------------+------+---------+

四、UDAF 函数案例

1、数据准备

zhangsan,89
lisi,67
wangwu,55
zhaoliu,78
tianqi,92

2、建立数据表

createtableifnotexists example_03(
    name string,
    score int)row format delimited
fieldsterminatedby',';

3、代码编写

需求：通过表中的数据，统计出学生的人数、总成绩、最高成绩、最低成绩、平均成绩、及格率

# -*- coding: utf-8 -*-# @Author   : 大数据章鱼哥# @Company  : 北京千锋互联科技有限公司import sys

scores =[]for line in sys.stdin:
    scores.append(int(line.strip()))# 计算人数
count =len(scores)# 计算总成绩
sum_score =sum(scores)# 计算最高成绩
max_score =max(scores)# 计算最低成绩
min_score =min(scores)# 计算平均成绩
avg_score = sum_score / count
# 计算及格率
rate =len(list(filter(lambda x: x >=60, scores)))/ count

# 输出最后的结果print('\t'.join(map(lambda x:str(x),[count, sum_score, max_score, min_score, avg_score, rate])))

4、应用函数

-- 添加文件
add file my_agg.py
-- 执行查询
select transform(*) using 'python3 my_agg.py' as (count, sum_score, max_score, min_score, avg_score, rate) from example_03

五、UDTF 函数案例

1、准备数据

zhangsan 78,89,92,96
lisi 67,75,83,94
王五 23,12

2、建立数据表

createtableifnotexists example_04(
name string,
scores array<String>)row format delimited 
fieldsterminatedby' '
collection items terminatedby',';

3、 需求实现

需求：根据表中的数据，将每一个成绩展开，绑定给每一个名字

# -*- coding: utf-8 -*-# @Author   : 大数据章鱼哥# @Company  : 北京千锋互联科技有限公司import sys

for line in sys.stdin:
    fields = line.strip().split('\t')iflen(fields)!=2:continue# 提取名字
    name = fields[0]# 提取所有成绩
    scores = fields[1].strip('[').rstrip(']').split(',')for s in scores:print("\t".join([name, s]))

4、函数应用

-- 添加文件addfile my_explode.py
-- 执行查询select transform(*)using'python3 my_explode.py'as(name, score)from example_04