{"id":32,"date":"2011-12-31T01:30:55","date_gmt":"2011-12-31T01:30:55","guid":{"rendered":"http:\/\/oren.lederman.name\/?p=32"},"modified":"2012-01-07T09:00:23","modified_gmt":"2012-01-07T09:00:23","slug":"analytic-functions-in-hadoop-hive-part-2","status":"publish","type":"post","link":"http:\/\/oren.lederman.name\/?p=32","title":{"rendered":"Analytic functions in Hadoop Hive part 2"},"content":{"rendered":"

 <\/p>\n

On this post I will explain how to write \u00a0a row_number() function that works with partitions. I assume that you have read Part 1 and loaded the sample data set. We will be using a small sample of original data set:<\/p>\n

[crayon lang=”sql” url=”http:\/\/oren.lederman.name\/wp-content\/uploads\/2011\/12\/post2_query0.txt”][\/crayon]<\/p>\n

For the\u00a0purpose\u00a0of this post, we will implement a UDF imitates the\u00a0behavior\u00a0of the following SQL statement:<\/p>\n

[crayon lang=”sql” url=”http:\/\/oren.lederman.name\/wp-content\/uploads\/2011\/12\/post2_query1.txt”][\/crayon]
\nReturning:
\n[csv2table source=”http:\/\/oren.lederman.name\/wp-content\/uploads\/2011\/12\/post2_query1.csv”]<\/p>\n

If we did not have to deal with partitions, all we had to do is keep a counter and increase it on each call to evaluate().\u00a0You can see such an example in \u00a0Hive’s contrib code (UDFRowSequence<\/a>). Note that this code assumes a single instance of UDFRowSequence (or else the same row number is assigned to more than one record).\u00a0However, using analytic functions with partitions is one of the most powerful features of analytic functions.<\/p>\n

The trick for writing an analytic function that supports partitioning is to sort<\/strong>\u00a0the data before calling the UDF. By sorting the records by the partitioning key (station_no<\/em> in our example) we can assure that all the records belonging to the same partition arrives\u00a0sequentially so we can reset the counter every time the partitioning key changes.\u00a0By using a\u00a0secondary\u00a0sort based on the order by keys (sample_date<\/em>\u00a0in our example) we assure that the records within each partition arrives in the order in which we want to number the records.<\/p>\n

This can be done using a SORT BY clause. SORT BY controls the order in which records arrive to a Reducer (read The difference between Sort By and Order By<\/a> for further information), which is exactly what we need here.
\n[crayon lang=”sql” url=”http:\/\/oren.lederman.name\/wp-content\/uploads\/2011\/12\/post2_query2.txt”][\/crayon]<\/p>\n

You probably noticed the DISTRIBUTE BY clause in the last example. Since Hive can choose to use more\u00a0\u00a0than one reducer (and if you process a large data set, this is what you want), we must make sure that all records of the same partition key reaches the same Reducer.<\/p>\n

Now that the data is sorted and distributed properly, the UDF code itself is quite simple. Note how evaluate() \u00a0saves the list of partition keys, and resets the counter every time the partition changes.<\/p>\n

[crayon url=”http:\/\/oren.lederman.name\/wp-content\/uploads\/2011\/12\/PurswayRowNumberUDF.txt”][\/crayon]<\/p>\n

Disclosure – the code in this example is based on the code of TiaraUDFRank<\/a>. I decided to use StringBuilder and rely on Text.toString() instead of arg.getBytes() as there seem to be a\u00a0bizarre\u00a0memory leak in the original code that returned wrong values.<\/p>\n

Related links<\/p>\n