使用Hive查询计算Hadoop中成功记录之间的差异

Question

SELECT * FROM mytable SORT BY customer_id, call_time;

Customer1    timestamp11
Customer1    timestamp12
Customer1    timestamp13
Customer2    timestamp21
Customer3    timestamp31
Customer3    timestamp32
...

Customer1    timestamp12-timestamp11
Customer1    timestamp13-timestamp12
Customer3    timestamp32-timestamp31
...

我试图从sql解决方案中调整解决方案，但我仍然受到配置单元的限制：它只接受from中的子查询，并且joins必须只包含相等项。

谢谢你。

编辑1：

public class DeltaComputerUDF extends UDF {
private String previousCustomerId;
private long previousCallTime;

public String evaluate(String customerId, LongWritable callTime) {
    long callTimeValue = callTime.get();
    String timeDifference = null;

    if (customerId.equals(previousCustomerId)) {
        timeDifference = new Long(callTimeValue - previousCallTime).toString();
    }

    previousCustomerId = customerId;
    previousCallTime = callTimeValue;

    return timeDifference;
}}

 SELECT customer_id, call_time, delta(customer_id, call_time) FROM mytable DISTRIBUTE BY customer_id SORT BY customer_id, call_time;

我可以在使用该函数之前对表数据进行排序，但我对此不满意，因为我希望避免这是一个开销。

第二：在分布式Hadoop配置的情况下，数据在可用的作业跟踪器之间进行分割。所以我相信这个函数会有多个实例，每个映射器有一个，所以在两个映射器之间有相同的客户数据拆分是可能的。在这种情况下，我将失去客户电话，这是不可接受的。

我不知道如何解决这个问题。我知道distribution BY确保所有具有特定值的数据都被发送到相同的reducer（从而确保SORT按预期工作），有人知道映射器是否有类似的东西吗？

流通常不如编码可比较的UDF或InputFormat对象的效率高。序列化和反序列化数据以将其传入和传出管道的效率相对较低。统一调试整个程序也更难。但是，它对于快速原型化和利用不是用Java编写的现有代码非常有用。对于不想编写Java代码的Hive用户来说，这可能是一种非常有效的方法。

因此，很明显，自定义脚本不是效率方面的最佳解决方案。

但是，我应该如何保持我的UDF功能，但确保它在分布式Hadoop配置中按预期工作呢？我在语言手册UDF wiki页面的UDF内部部分找到了这个问题的答案。如果我编写查询：

 SELECT customer_id, call_time, delta(customer_id, call_time) FROM (SELECT customer_id, call_time FROM mytable DISTRIBUTE BY customer_id SORT BY customer_id, call_time) t;

（抱歉没有添加链接，但我没有足够的信誉点，所以不允许我这样做）

Answer 1

这是一个老问题，但为了将来的参考，我在这里写了另一个命题：

配置单元窗口函数允许在查询中使用上一个/下一个值。

类似的代码查询可能是：

SELECT customer_id, call_time - LAG(call_time, 1, 0) OVER (PARTITION BY customer_id ORDER BY call_time) FROM mytable;