Python实现UDF函数的逻辑

　　Hive中使用Python自定义函数的时候，其实是一个重定向输入的过程。将表中的指定的列逐行读取，我们在Python的脚本中就可以依次对每一行的数据进行处理。

　　import sys

　　# sys.stdin: 标准输入流，我们使用的input()函数，其实就是stdin的一个方法。

　　# hive执行自定义函数的时候，会逐行读取指定列的内容，我们就可以使用循环依次处理读取到的每一行的数据。

　　for line in sys.stdin:

　　# 读取到的这一行的数据，其中包含了若干个列，列于列之间以\t进行分隔

　　# 切割出来每一个列

　　columns = line.strip().split('\t')

　　# 对列进行处理

　　# 处理之后的结果，使用print直接打印即可

　　# 如果需要返回多个列，多个列之间需要使用\t进行分隔

　　print('hello world')

　　在使用UDF的时候，需要按照如下的步骤

　　# 1. 加载脚本文件

　　add file <脚本文件>

　　# 2. 执行脚本，执行Python的UDF函数的时候，只能使用transform函数

　　select transform(<字段>) using 'python3 <脚本文件>' as (<字段列表>) from <表>

　　自定义UDF函数案例

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

# 1. 准备数据文件
zhangsan,male,19
lisi,female,20
wangwu,male,18
zhaoliu,female,19

# 2. 创建数据表
create table if not exists example_01(
t_name string,
t_gender string,
t_age int
)
row format delimited
fields terminated by ',';

　　需求：将所有的列都转成大写

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

# string_upper.py
import sys

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

-- 应用UDF
add file 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;

　　需求：只需要将姓名转成大写

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

# string_name_upper.py
import sys

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

# 3. 打印处理之后的结果
print('\t'.join([name, gender, age]))

-- 查询所有字段，将名字转成大写
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. 准备数据表
create table if not exists example_02(
username string,
idcard string
)
row format delimited
fields terminated by ',';

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

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

# 身份证验证
# 如果身份证不是一个合法的身份证，输出空
# 如果身份证是合法的，解析出年龄、生日、性别

import sys
import re
import datetime

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

for line in sys.stdin:
# 1. 切割出来每一个组成部分
fields = line.strip().split('\t')
# 2. 如果长度不到两位，说明数据有问题，不做任何处理
if len(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 is None:
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 = '男' if int(m.group('gender')) % 2 != 0 else '女'
# 拼接生日
birthday = f'{year}-{month}-{day}'
print('\t'.join([name, id_card, birthday, str(age), gender]))

-- 添加文件
add file 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 | 女 |
+-------+---------------------+-------------+------+---------+

　　案例三：自定义聚合函数

# 准备数据
zhangsan,89
lisi,67
wangwu,55
zhaoliu,78
tianqi,92

# 建立数据表
create table if not exists example_03(
name string,
score int
)
row format delimited
fields terminated by ',';

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

　　# -*- 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])))

　　案例四：自定义展开函数

　　# 1. 准备数据

　　zhangsan 78,89,92,96

　　lisi 67,75,83,94

　　王五 23,12

　　# 2. 建立数据表

　　create table if not exists example_04(

　　name string,

　　scores array

　　)

　　row format delimited

　　fields terminated by ' '

　　collection items terminated by ','

　　;

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

　　# -*- coding: utf-8 -*-

　　# @Author : 大数据章鱼哥

　　# @Company : 北京千锋互联科技有限公司

　　import sys

　　for line in sys.stdin:

　　fields = line.strip().split('\t')

　　if len(fields) != 2:

　　continue

　　# 提取名字

　　name = fields[0]

　　# 提取所有成绩

　　scores = fields[1].strip('[').rstrip(']').split(',')

　　for s in scores:

　　print("\t".join([name, s]))