Hive中使用Python文件实现UDF

背景：表中有一个降水强度字段，但是并不是真实的降雨量，后来数据模型的同事给了一个python脚本，按照脚本可以大体转换成降雨量。
思路：
1.想过用 python 现将表中的数查出来放到文件中，然后在用给的python逻辑对降水强度字段进行处理，最后在加载进表中。 但是这样太麻烦了。
2.也想过用给的python脚本查出一些数值，然后建立一个数学模型，找出一个计算关系。最后因为自己非数学专业，也不是专业分析人员，放弃了该想法。
3.最后，突然想到，是不是能像 udf 函数一样，将这段python 代码，也当做一个 udf 函数嵌到 sql 中呢。
看来解决问题，思路还是最重要，什么样的思路决定了解决方法，决定了最后的难易程度。

这是我原本的 sql，现在 intensity 字段需要用给定的python逻辑进行一个加工。

select
    cityid,
    city_name,
    di_cityid,
    substr(server_time1,1,14) as server_time1,
    intensity as intensity,
    temperature as temperature,
    round((wind_speed / 36),2) as wind_speed
    from tablename
    where concat(year,month,day) = '${DT}'
    and di_cityid in (9,11,17,47,95)
    and intensity < 1

这是拿到的转换代码。我在里面加了些注释。

def intense2dbz(intn):
    val = (intn + 0.15) * 16.0 * 5
    # 当val>70，return 70*val 。 当val <= 70，返回val * val 因为True 和 False 在python 计算中就代表 1 和 0
    return 70 * (val > 70) + val * (val <= 70)
 
def radar2precip_metric(radar):
    dbz = intense2dbz(radar)
    # power(a,b) --> a的b次方
    val = np.power(np.power(10.0, dbz / 10.0) / 200, 5.0 / 8.0)
    val = val - 0.2051
    return val * (val > 0)

ok，下面是根据给的转换方法写的 udf 函数代码：

# -*- coding:utf-8 -*-
import numpy as np
import sys

def intense2dbz(intn):
    val = (intn + 0.15) * 16.0 * 5
    return 70 * (val > 70) + val * (val <= 70)
 
def radar2precip_metric(radar):
    dbz = intense2dbz(radar)
    val = np.power(np.power(10.0, dbz / 10.0) / 200, 5.0 / 8.0)
    val = val - 0.2051
    return val * (val > 0)

if __name__ == "__main__":
    #循环读取每一行数据。
    for line in sys.stdin:
        # 去掉每一行后面的 \n
        line = line.strip()
        # 这里应该要把所有的字段都列出来，因为你读了一行的数据，其实包含了所有的字段。split('\t')应该要源表的分隔符一致。
        cityid,city_name,di_cityid,server_time1,intensity,temperature,wind_speed = line.split("\t")
        # 相关字段进行相关操作。
        new_server_time1 = server_time1[0:14]
        new_wind_speed = round((float(wind_speed) / 36),2)
        new_intensity = abs(radar2precip_metric(float(intensity)))        
        #最后输出新的所有的字段。
        print("\t".join([cityid,city_name,didi_cityid,new_server_time1,str(new_intensity),temperature,str(new_wind_speed)]))

下面是在 hive 里使用上述文件：

-- 首先是添加文件，java需要一个jar包。python就是一个py文件。 我这是hdfs 上面的相对路径。绝对路径也可以。
add file ./intensity.py;

select
transform(cityid,city_name,di_cityid,server_time1,intensity,temperature,wind_speed)
using 'python intensity.py'
as (cityid,city_name,di_cityid,server_time1,intensity,temperature,wind_speed)
from tablename
where concat(year,month,day) = '${DT}'
and di_cityid in (9,11,17,47,95)
and intensity < 1

transform 是关键字，将原本的字段全都放进去，然后 using 添加后的 python文件，最后返回新的字段。
这里，传递进去的字段的数量，和最后生成的字段数量不一定相同。可能一个字段在python脚本中分解成多个字段，也可能多个字段最后加工成了一个字段。
但是如果想：
select cityid,city_name,transform(intensity) using 'python intensity.py' as intensity,.....
这样我试过，是不行的。

另外，在实际操作中发现了两个问题：
1.最后python文件返回必须是 str 类型，所有我在代码中，都强转了 str 类型。
2.承接上一条，使用了python文件后，你的字段类型也就变成了 str 类型。这时候在求最大值这种需要注意，你还要在转回数值类型，否则求出来的最大值就不准确。貌似hive 没法自动识别出来你这个字符串类型的字段中，实际上都是数值类型的值。

第一次使用 python 文件嵌入到hive 中使用，如果不是碰到这个问题，还不知道有这种操作。欠缺的地方还很多，欢迎各位指正，期待大神指教一二。