对您来说，最简单的事情可能是让Map器以id作为键输出数据，这将保证一个reducer将获得特定id的所有记录，然后在reducer阶段进行处理。
例如，
输入数据：

b1  2:00pm z1   1
 b2  3:00pm z2   2
 c1  4:00pm y2   1
 b3  3:00pm z3   3
 c4  4:00pm x2   2

Map程序代码：

# !/usr/bin/env python

import sys
for line in sys.stdin:
    line = line.strip()
    cols = line.split("\t")
    key = cols[-1]
    print key + "\t" + line

Map输出：

1  b1  2:00pm z1   1
 2  b2  3:00pm z2   2
 1  c1  4:00pm y2   1
 3  b3  3:00pm z3   3
 2  c4  4:00pm x2   2

减速器1输入：

1  b1  2:00pm z1   1
 1  c1  4:00pm y2   1

减速器2输入：

2  b2  3:00pm z2   2

减速器3输入：

3  b3  3:00pm z3   3

减速机代码：

# !/usr/bin/env python

import sys
for line in sys.stdin:
    line = line.strip()
    cols = line.split("\t")
    orig_line = "\t".join(cols[1:])
    # do stuff...

请注意，这样一个reducer可能会获得多个键，但数据将被排序，您可以使用mapred.reduce.tasks选项控制reducer的数量。
编辑如果你想收集你的数据在减速器每键你可以这样做（不确定它会运行，但你得到的想法）

# !/usr/bin/env python

import sys
def process_data(key_id, data_list):
   # data_list has all the lines for key_id

last_key = None
data = []
for line in sys.stdin:
    line = line.strip()
    cols = line.split("\t")
    key = cols[0]
    if last_key and key != last_key:
        process_data(last_key, data)
        data = []
    orig_line = "\t".join(cols[1:])
    data.append(orig_line)
    last_key = key
process_data(last_key, data)

如果您不担心在reducer步骤中内存不足，可以将代码简化为：

# !/usr/bin/env python

import sys
from collections import defaultdict
def process_data(key_id, data_list):
   # data_list has all the lines for key_id

all_data = defaultdict(list)
for line in sys.stdin:
    line = line.strip()
    cols = line.split("\t")
    key = cols[0]
    orig_line = "\t".join(cols[1:])
    all_data[key].append(orig_line)
for key, data in all_data.iteritems():
    process_data(key, data)

如果我理解您的问题，最好的方法是将数据集加载到配置单元表中，然后用python编写udf。在那之后，像这样做：

select your_python_udf(name, time, dept, id) from table group by id;

这看起来像是reduce阶段，所以在启动查询之前可能需要这个

set mapred.reduce.tasks=50;

如何创建自定义自定义项：
Hive插件
创建函数