HadoopコミュのHiveのUDFについて

Hiveに関する質問がこのコミュニティに相応しいのかわかりませんが、ご存じの方がいらしたらお願いします。
今、ユーザIDを含むデータが大量にあり、ユーザごとに独立した、レコード事にユニークなシリアル番号を付与するUDFを作っています。

例えば、
datatableというテーブルに
user1, 20101104
user1, 20101105
user2, 20101104
user2, 20101105
というデータがあれば、
1, user1, 20101104
2, user1, 20101105
1, user2, 20101104
2, user2, 20101105
というふうに、頭にシリアル番号を付与するためのUDFです。

例えばHiveQLにて
select serial(user_id), user_id, log_date from datatable distribute by user_id sort by user_id, log_date;
と実行することで、distribute byとsort byでユーザ順に並び替えた後にReducer処理でUDFを呼び出すことを考えています。しかし、普通にselect内でUDFを呼び出すだけでは、Mapper側でUDFが呼び出されてしまっているようで、Reducer側で呼び出す方法がわかりません。このため、同一ユーザが同一のシリアル番号のデータを持ってしまうという問題が発生しております。

＜可能性のある出力結果＞
1, user1, 20101104
1, user1, 20101105
2, user2, 20101104
2, user2, 20101105

今は回避策として以下のHiveQLにて、2回selectを行い、1回目でdistribute byとsort byを行い、2回目でUDFを使うようにしていますが、1回目の分割結果が2回目に正しく反映される保証があるのかどうか分からないのと、2回MapReduceを実行してしまうため非効率だという点が問題です。

select serial(t1.user_id), t1.* from (select user_id, log_date from datatable distribute by user_id sort by user_id, log_date) t1;

＃　Hiveを使わずに直接MapReduceを使えばたわいもないことなのですが、Hiveと組み合わせる必要があるのです。
どのようにしたら、Reducer側でUDFを実行できるのでしょうか？
もしHiveに精通している方がいらっしゃれば、お力を拝借させてください。
よろしくお願いします。

参考までに、作成したUDFを添付します。

public class RowSequencer extends UDF {
private final Map<String, LongWritable> counterMap;
public RowSequencer() {
counterMap = new HashMap<String,LongWritable>();
}
public LongWritable evaluate(Text exp) {
LongWritable result = counterMap.get(exp.toString());
if (result == null) {
result = new LongWritable(0);
counterMap.put(exp.toString(), result);
}
result.set(result.get() + 1);
return result;
}
}