我正在尝试按组计算记录之间的差异,并按组包括行号。这可以通过使用窗口函数在HIVE中使用滞后和行号函数来实现。尝试使用PIG和pythonUDF重新创建它。
在下面的示例中,我需要每个名称的行号从1重新开始,并为新的月份(新记录)增加行号。此外,我需要每个名字的余额与上月的差额。
输入数据
name month balance
A 1 10
A 2 5
A 3 15
B 2 20
B 3 10
B 4 45
B 5 50
输出数据
name month balance row_number balance_diff
A 1 10 1 0
A 2 5 1 -5
A 3 15 3 10
B 2 20 1 0
B 3 10 2 -10
B 4 45 3 35
B 5 50 4 5
如何使用PIG和python UDF来实现这一点?以下是我尝试过的。
PIG
output = foreach (group input by (name)) {
sorted = order input BY month asc;
row_details= myudf.rownum_and_diff(sorted.(month, balance));
generate flatten (sorted), flatten (row_details));
};
Python UDF
def row_num(mth):
return [x+1 for x,y in enumerate (mth)]
def diff(bal, n=1):
return [x-y if (x is not None and y is not None) else 0.0
for x,y in zip(bal, [:n] + bal)]
@outputSchema('udfbag:bag{udftuple:tuple(row_number: int, balance_diff: int)}')
def row_metrics(mthbal):
mth, bal = zip(*mthbal)
row_number = row_num(mth)
balance_diff = diff(bal)
return zip(row_number, balance_diff)
我的python函数可以工作。然而,一旦我将结果输入PIG,我就很难组合这两个袋子(sorted和row_detail)。非常感谢您的帮助。
我还看到PIG中的enumerate函数对行号执行我想要的操作。然而,作为学习PIG的一部分,我正在寻找一个使用python UDF的解决方案。
试试这个。
Python UDF:
def row_num(mth):
return [x+1 for x,y in enumerate (mth)]
def diff(bal, n=1):
return [0]+[x-y for x,y in zip(bal[n:],bal[:-n])]
@outputSchema('udfbag:bag{udftuple:tuple(name: chararray, mth: int, row_number: int, balance_diff: int)}')
def row_metrics(mthbal):
name, mth, bal = zip(*mthbal)
row_number = row_num(mth)
balance_diff = diff(bal)
return zip(name,mth,row_number, balance_diff)
Pig脚本:
register 'myudf.py' using jython as myudf;
inpdat = load 'input.dat' using PigStorage(',') as (name:chararray, month:int, balance:int);
outdat = foreach (group inpdat by name) {
sorted = order inpdat BY month asc;
row_details = myudf.row_metrics(sorted);
generate flatten (row_details);
};
dump outdat;
在我的案例中,使用piggybank中的缝合功能是有效的。有兴趣学习其他方法。
REGISTER /mypath/piggybank.jar;
define Stitch org.apache.pig.piggybank.evaluation.Stitch;
input = load 'input.dat' using PigStorage(',') as (name:chararray, month:int, balance:int);
output = FOREACH (group input by name) {
sorted = ORDER input by month asc;
udf_fields = myudf.row_metrics(sorted.(month, balance));
generate flatten(Stitch(sorted,udf_fields)) as (name, month, balance, row_number, balance_diff);
};