title: Compute differences between succesive records in Hadoop with Hive Queries
body: <p>I have a Hive table that holds data of customer calls.
  For simplicity consider it has 2 columns, first column holds the customer ID and the second column holds the timestamp of the call (unix timestamp).</p>
  
  <p>I can query this table to find all the calls for each customer:</p>
  
  <pre><code>SELECT * FROM mytable SORT BY customer_id, call_time;
  </code></pre>
  
  <p>The result is:</p>
  
  <pre><code>Customer1    timestamp11
  Customer1    timestamp12
  Customer1    timestamp13
  Customer2    timestamp21
  Customer3    timestamp31
  Customer3    timestamp32
  ...
  </code></pre>
  
  <p>Is it possible to create a Hive query that returns, for each customer, starting from the second call, the time interval between two succesive calls?
  For the above example that query should return:</p>
  
  <pre><code>Customer1    timestamp12-timestamp11
  Customer1    timestamp13-timestamp12
  Customer3    timestamp32-timestamp31
  ...
  </code></pre>
  
  <p>I have tried to adapt the solutions from the <a href="https://stackoverflow.com/questions/634568/how-to-get-difference-between-two-rows-for-a-column-field">sql solution</a>, but I'm stuck with the Hive limitations: <a href="https://cwiki.apache.org/confluence/display/Hive/LanguageManual+SubQueries" rel="nofollow noreferrer">it accepts subqueries only in FROM</a> and <a href="https://cwiki.apache.org/confluence/display/Hive/LanguageManual+Joins" rel="nofollow noreferrer">joins must contain only equalities</a>.</p>
  
  <p>Thank you.</p>
  
  <p><strong>EDIT1:</strong></p>
  
  <p>I have tried to use a Hive UDF function:</p>
  
  <pre><code>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;
  }}
  </code></pre>
  
  <p>and use it with the name "delta".</p>
  
  <p>But it seems (from the logs and result) that it is being used at MAP time. 2 problems arise from this:</p>
  
  <p><strong><em>First:</em></strong> The table data must be sorted by Customer ID and timestamp BEFORE using this function. The query:</p>
  
  <pre><code> SELECT customer_id, call_time, delta(customer_id, call_time) FROM mytable DISTRIBUTE BY customer_id SORT BY customer_id, call_time;
  </code></pre>
  
  <p>does not work because the sorting part is performed at REDUCE time, long after my function is being used.</p>
  
  <p>I can sort the table data before using the function, but I'm not happy with this because it is an overhead I hope to avoid.</p>
  
  <p><strong><em>Second:</em></strong> In case of a distributed Hadoop configuration, data is split among the available job trackers. So I believe there will be multiple instances of this function, one for each mapper, so it is possible to have the same customer data split between 2 mappers. In this case I will lose customer calls, which is not acceptable.</p>
  
  <p>I don't know how to solve this issue. I know that DISTRIBUTE BY ensures that all data with a specific value is sent to the same reducer (thus ensuring that SORT works as expected), does anybody know if there is something similar for the mapper?</p>
  
  <p>Next I plan to follow libjack's suggestion to use a reduce script. This "computation" is needed between some other hive queries, so I want to try everything Hive offers, before moving to another tool, as suggested by Balaswamy vaddeman.</p>
  
  <p><strong>EDIT2:</strong></p>
  
  <p>I started to investigate the custom scripts solution. But, in the first page of chapter 14 in Programming Hive book (this chapter presents the custom scripts), I found the following paragraph:</p>
  
  <blockquote>
    <p>Streaming is usually less efficient than coding the comparable UDFs or
    InputFormat objects. Serializing and deserializing data to pass it in and
    out of the pipe is relatively inefficient. It is also harder to debug the whole
    program in a unified manner. However, it is useful for fast prototyping
    and for leveraging existing code that is not written in Java. For Hive
    users who don’t want to write Java code, it can be a very effective
    approach.</p>
  </blockquote>
  
  <p>So it was clear that the custom scripts is not the best solution in terms of efficiency.</p>
  
  <p>But how should I keep my UDF function, but make sure it works as expected in a distributed Hadoop configuration? I found the answer to this question in the UDF Internals section of the Language Manual UDF wiki page. If I write my query:</p>
  
  <pre><code> 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;
  </code></pre>
  
  <p>it is executed at REDUCE time and DISTRIBUTE BY and SORT BY constructs guarantee that all the records from the same customer are being processed by the same reducer, in order of calls.</p>
  
  <p>So the above UDF and this query construct solve my problem.</p>
  
  <p>(Sorry for not adding the links, but I'm not allowed to do it because I don't have enough reputation points)</p>