使用Python實現(xiàn)HIVE的UDF函數(shù)
在處理一些復(fù)雜邏輯時候,python這種面向過程的語言相比于SQL更符合人的思維方式。相信有不少同學(xué)曾經(jīng)感慨,如果能用python處理數(shù)據(jù)庫中的數(shù)據(jù)就好了。那么今天它來了。
首先用python寫處理復(fù)雜邏輯的自定義的函數(shù)(一陽指),再將函數(shù)代碼嵌入SQL(獅吼功)就能合并成了一整招:UDF
下面我用一個栗子來說明一些兩者處理數(shù)據(jù)過程中的差異,在介紹栗子之前,先介紹一些with as。與python 創(chuàng)建函數(shù)或者類一樣,with as 用于創(chuàng)建中間表
簡單來做個介紹
- select
- *
- from(select * from table where dt='2021-03-30')a
可以寫成
- with a as (select * from table where dt='2021-03-30' )
- select * from a
簡單的SQL看不出這樣的優(yōu)勢(甚至有點多此一舉),但是當(dāng)邏輯復(fù)雜了之后我們就能看出這種語法的優(yōu)勢,他能從底層抽取中間表格,讓我們只專注于當(dāng)前使用的表格,進而可以將復(fù)雜的處理邏輯分解成簡單的步驟。
如下面地表格記錄了用戶適用app過程中每個行為日志地時間戳,我們想統(tǒng)計一下用戶今天用了幾次app,以及每次的起始時間和結(jié)束時間是什么時候,這個問題怎么解呢?
SQL實現(xiàn)方式
首先用with as 構(gòu)建一個中間表(注意看on 和 where條件)
- with t1 as
- (select
- x.uid,
- case when x.rank=1 then y.timestamp_ms
- else x.timestamp_ms
- end as start_time,
- case when x.rank=1 then x.timestamp_ms
- else y.timestamp_ms end as end_time
- from
- (select
- uid,
- timestamp_ms,
- row_number()over(partition by uid order by timestamp_ms) rank
- from tmp.tmpx) x
- left outer join
- (select
- uid,
- timestamp_ms,
- row_number()over(partition by uid order by timestamp_ms) rank
- from tmp.tmpx) y
- on x.uid=y.uid and x.rank=y.rank-1
- where x.rank=1 or y.rank is null or y.timestamp_ms-x.timestamp_ms>=300)
首先我們用開窗函數(shù)錯位相減,用where條件篩選出我們需要的列,其中
x.rank=1 抽取出第一行
y.rank is null 抽取最后一樣
y.timestamp_ms-x.timestamp_ms>=300抽取滿足條件的行,如下:
當(dāng)然這個結(jié)果并不是我們要的結(jié)果,需要將上述表格中某一行數(shù)據(jù)的end-time和下一條數(shù)據(jù)的start-time結(jié)合起來起來,構(gòu)造出時間段
好的,按照上面我們所說的那么下面我們不用關(guān)心底層的邏輯,將注意力專注于這張中間表t1
- select
- a.uid,end_time as start_time,start_time as end_time
- from
- (select uid,start_time,row_number()over(partition by uid order by start_time) as rank from t1) a
- join
- (select uid,end_time,row_number()over(partition by uid order by end_time) as rank from t1)b
- on
- a.uid=b.uid and a.rank=b.rank+1
同樣,排序后錯位相減,然后就可以打完收工了~
UDF實現(xiàn)方式
首先我們假設(shè)上述數(shù)據(jù)存儲在csv中,
用python 處理本地文件data.csv,按照python的處理方式寫代碼(這里就不一句句解釋了,會python的同學(xué)可以跳過,不會的同學(xué)不妨自己動手寫一下)
- def life_cut(files):
- f=open(files)
- act_list=[]
- act_dict={}
- for line in f:
- line_list=line.strip().split()
- key=tuple(line_list[0:1])
- if key not in act_dict:
- act_dict.setdefault(key,[])
- act_dict[key].append(line_list[1])
- else:
- act_dict[key].append(line_list[1])
- for k,v in act_dict.items():
- k_str=k[0]+"\t"
- start_time = v[0]
- last_time=v[0]
- i=1
- while i<len(v)-1:
- if int(v[i])-int(last_time)>=300:
- print(k_str+"\t"+start_time+"\t"+v[i-1])
- start_time=v[i]
- last_time = v[i]
- i=i+1
- else:
- last_time = v[i]
- i=i+1
- print(k_str+"\t"+start_time+"\t"+v[len(v)-1])
- # print(k_str + "\t" + start_time + "\t" + v[i])
- if __name__=="__main__":
- life_cut("data.csv")
得到結(jié)果如下:

那么下面我們將上述函數(shù)寫成udf的形式:
- #!/usr/bin/env python
- # -*- encoding:utf-8 -*-
- import sys
- act_list=[]
- act_dict={}
- for line in sys.stdin:
- line_list=line.strip().split("\t")
- key=tuple(line_list[0:1])
- if key not in act_dict:
- act_dict.setdefault(key,[])
- act_dict[key].append(line_list[1])
- else:
- act_dict[key].append(line_list[1])
- for k,v in act_dict.items():
- k_str=k[0]+"\t"
- start_time = v[0]
- last_time=v[0]
- i=1
- while i<len(v)-1:
- if int(v[i])-int(last_time)>=300:
- print(k_str+"\t"+start_time+"\t"+v[i-1])
- start_time=v[i]
- last_time = v[i]
- i=i+1
- else:
- last_time = v[i]
- i=i+1
- print(k_str+"\t"+start_time+"\t"+v[len(v)-1])
這個變化過程的關(guān)鍵點是將 for line in f 替換成 for line in sys.stdin,其他基本上沒什么變化
然后我們再來引用這個函數(shù)
先add這個函數(shù)的路徑add file /xxx/life_cut.py 加載udf路徑,然后再使用
- select
- TRANSFORM (uid,timestamp_ms) USING "python life_cut.py" as (uid,start_time,end_time)
- from tmp.tmpx
總結(jié)
從上述案例我們可以看出,
UDF和SQL的區(qū)別在于,在處理復(fù)雜邏輯時候,UDF相比SQL能更高效地組織起來邏輯并落地實現(xiàn)功能。UDF和普通腳本的關(guān)鍵區(qū)別所在在于將 for line in f 替換成 for line in sys.stdin,常規(guī)函數(shù)一般是將文件一行行讀入,UDF是從標(biāo)準(zhǔn)輸入一行行加載數(shù)據(jù)。希望大家平時沒事的時候好好練練python,切莫書到用時方恨少。