hive结合python： Transform的使用

huashan8

　　使用场景是这样的：
　　如何将一行变换成多行，如下，第一列的值是多个以分号分隔的属性：属性值字符串，现在我想把这列拆     成属性 和 属性值两列来显示，如下：

# ["规格:RN1-10/51;规格:RN1-10/52;规格:RN1-10/53", '11', '22']
# ["规格", "RN1-10/51", '11', '22']
# ["规格", "RN1-10/52", '11', '22']
# ["规格", "RN1-10/53", '11', '22']
　　在hive中可以使用Transform来完成这样的数据处理场景，注意一下的语法格式
　　Transform中的值作为输入， 然后传递给python脚本，最后经过python的处理后，将原来的三列转换成
　　四列进行显示

select
TRANSFORM(p.joint_attr_values, p.collect_product_id, p.released_id)
USING 'python split_product_attrs.py'
as (custom_attr , custom_attr_value, collect_product_id, released_product_id)
　　还有一点需要注意：

select
TRANSFORM(p.joint_attr_values, p.collect_product_id, p.released_id)
USING 'python split_product_attrs.py'
as (custom_attr , custom_attr_value, collect_product_id, released_product_id)
from
(  这里应该是另外一个select语句，用于Transform的输入，最好是一一对应的，否则会出错  )

　　另外在调用这个python脚本的时候，一定要先把源给加上，语法如下
　　hive> add file '你的python文件路径'
　　完整例子：

select
TRANSFORM(p.joint_attr_values, p.collect_product_id, p.released_id)
USING 'python split_product_attrs.py'
as (custom_attr , custom_attr_value, collect_product_id, released_product_id)
from
(select pc.joint_attr_values, pm.collect_product_id, pm.released_id from
products_compared_${hiveconf:wid_version} pc
left join products_merged_${hiveconf:wid_version} pm
on pc.collect_id = pm.collect_product_id
where pm.released_id is not null
and pc.joint_attr_values is not null and pc.joint_attr_values != '' and pc.joint_attr_values != 'null') as p ;  

　　下面是python的脚本，用于将三列转换为四列，这里就比较简单了，主要用于测试，代码随便写了一下

#!/usr/bin/python
# #_*_ coding: utf-8 _*_
import sys
import datetime
# "规格:RN1-10/50;规格:RN1-10/50;规格:RN1-10/50"
# ["规格:RN1-10/51;规格:RN1-10/52;规格:RN1-10/53", '11', '22']
# ["规格", "RN1-10/51", '11', '22']
# ["规格", "RN1-10/52", '11', '22']
# ["规格", "RN1-10/53", '11', '22']
for line in sys.stdin:
values = line.split('\t')
values = [ i.strip() for i in values ]
tmp = values[0]
key_values = tmp.split(";")
for kv in key_values:
k = kv.split(":")[0]
v = kv.split(":")[1]
print '\t'.join([k,v,values[1],values[2]])

下面是网上找个一个列子，跟我这个差不多


hive 使用streaming 的map 脚本来处理数据

 



在对hive 进行 select 查询的时候 我们可以编写 python 、php 、c++等脚本来进行相应的数据处理，我们要用到hive 的 TRANSFORM 和 using
看例子：

add file /www/FCCS_Data/ComETL/hive/sql_map/demo.py ;
from access_fccs select TRANSFORM (time) using 'python demo.py' where week=41 limit 10 ;

或者：

add file /www/FCCS_Data/ComETL/hive/sql_map/demo.py;
select TRANSFORM (time) using 'python demo.py' as (time) from (select * from access_fccs where week=41 limit 10) a ;

上面 SQL的意思是 将查询结果集中的 time 通过 demo.py 进行处理 返回相应的结果，至于都做什么处理了，请看下面的demo.py代码。
这里我们需要注意的是，一定要先 将map脚本添加到分布式缓存中，否则会报错 metadata.HiveException: [Error 20003]: An error occurred when trying to close the Operator running your custom script. at 这种错误
特别说明 这里的 路径为本地路径 不是分布式HDFS 路径，新手比较容易搞混
然后看看我们的demo.py 是怎么写的

#!/usr/bin/env python
#-*- coding:utf-8 -*-
import sys
import datetime
import time
#把时间戳转成字符串形式
def timestamp_toString(stamp):
return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(stamp))
for line in sys.stdin:
print timestamp_toString(float(line))

董python 的朋友不难看出 这里是将我们的时间戳转化成日期格式输出
因此我们执行完毕 hive SQL 后 的结果为：

Total MapReduce CPU Time Spent: 23 seconds 880 msec
OK
2013-10-11 13:11:47
2013-10-11 13:11:47
2013-10-11 13:11:47
2013-10-11 13:11:47
2013-10-11 13:11:47
2013-10-11 13:11:48
2013-10-11 13:11:48
2013-10-11 13:11:48
2013-10-11 13:11:48
2013-10-11 13:11:48
Time taken: 33.411 seconds