使用hive udf函数计算运行总和

Question

我是Hive的新手,我想事先赦免我的无知,以免下面的任何事情发生.我有一张表如下:

SELECT a.storeid, a.smonth, a.sales FROM table a;
1001    1       35000.0
1002    2       35000.0
1001    2       25000.0
1002    3       110000.0
1001    3       40000.0
1002    1       40000.0

我的目标输出如下:

1001    1       35000.0 35000.0
1001    2       25000.0 60000.0
1001    3       40000.0 100000.0
1002    1       40000.0 40000.0
1002    2       35000.0 75000.0
1002    3       110000.0 185000.0

我编写了一个简单的hive udf sum类来实现上述功能,并在查询中使用了SORT BY storeid,smonth:

SELECT a.storeid, a.smonth, a.sales, rsum(sales)
FROM (SELECT * FROM table SORT BY storeid, smonth) a;

显然,它不会产生上述输出,因为只有一个映射器,并且调用了相同的udf实例,它在总集合上生成一个运行总和.我的目标是为每个storeid重置udf类中的runningSum实例变量,以便evaluate函数返回上面的输出.我使用了以下内容:1.传递storeid变量rsum(sales,storeid),然后我们可以在udf类中正确处理这种情况.2.使用2个映射器,如以下查询中所示:

set mapred.reduce.tasks=2;
SELECT a.storeid, a.smonth, a.sales, rsum(sales)
FROM (SELECT * FROM table DISTRIBUTE BY storeid SORT BY storeid, smonth) a;

1002    1       40000.0 40000.0
1002    2       35000.0 75000.0
1002    3       110000.0 185000.0
1001    1       35000.0 35000.0
1001    2       25000.0 60000.0
1001    3       40000.0 100000.0

为什么1002总是出现在顶部？我想就除了上述方法之外我可以实现相同的其他方法(例如子查询/连接)寻求你的建议.此外,您建议的方法的时间复杂性是多少？

Answer 1

Hive提供了一种更好的方法来执行此操作 -
请按照以下过程来实现目标输出

创建一个可以包含您的数据集的配置单元表 -

1001    1       35000.0
1002    2       35000.0
1001    2       25000.0
1002    3       110000.0
1001    3       40000.0
1002    1       40000.0

现在只需在您的蜂巢终端中运行以下命令 -

SELECT storeid, smonth, sales, SUM(sales) OVER (PARTITION BY storeid ORDER BY smonth) FROM table_name;

输出将像 -

1001  1  35000.0  35000.0
1001  2  25000.0  60000.0
1001  3  40000.0  100000.0
1002  1  40000.0  40000.0
1002  2  35000.0  75000.0
1002  3  110000.0 185000.0

我希望这可以帮助您获得目标输出.

Answer 2

或者，您可以查看此Hive 票证，其中包含多个功能扩展。
其中有一个累积和实现（GenericUDFSum）。

该函数（称为“rsum”）采用两个参数，即 id 的哈希值（通过该哈希值在减速器之间对记录进行分区）以及要求和的相应值：

select t.storeid, t.smonth, t.sales, rsum(hash(t.storeid),t.sales) as sales_sum 
  from (select storeid, smonth, sales from sm distribute by hash(storeid) 
    sort by storeid, smonth) t;

1001  1  35000.0  35000.0
1001  2  25000.0  60000.0
1001  3  40000.0  100000.0
1002  1  40000.0  40000.0
1002  2  35000.0  75000.0
1002  3  110000.0 185000.0