Разделение ввода для функции Map в Hadoop

Самое простое для вас, вероятно, было бы сделать так, чтобы сопоставитель выдавал данные с идентификатором в качестве ключа, что гарантирует, что один редуктор получит все записи для определенного идентификатора, а затем выполнит вашу обработку на этапе редьюсера.

Например,

Входные данные:

 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

Код сопоставления:

#!/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

Вывод карты:

 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...

Обратите внимание, что таким образом один редьюсер может получить несколько ключей, но данные будут упорядочены, и вы можете контролировать количество редюсеров с помощью параметра mapred.reduce.tasks.

EDIT Если вы хотите собрать свои данные в редукторе для каждого ключа, вы можете сделать что-то вроде этого (не уверен, что он будет работать как есть, но вы поняли идею)

#!/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)

Если вас не беспокоит нехватка памяти на шаге редуктора, вы можете упростить код следующим образом:

#!/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)

Если я понял ваш вопрос, лучший способ — загрузить набор данных в таблицу hive, а затем написать UDF на python. После этого сделайте что-то вроде этого:

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

Это похоже на фазу сокращения, поэтому вам, возможно, это понадобится перед запуском запроса.

set mapred.reduce.tasks=50;

Как создать пользовательскую UDF:

Плагины Hive

Создать функцию