MapReduce программирования для фильтрации большого входного файла

Question

У меня есть очень большой входные текстовый файл в следующем формате:

ID \t time \t product \t Description \t Status 

Колонок Статуса ограничен содержать либо в нижнем регистре а, с, я или верхний случай А , S, I или смешанный из двух (элемент образца в статусе col: a, si, I, asi, ASI, aSI, Asi ...)

Что я хочу достичь, это отфильтровать этот выходной файл на основе состояния с использованием MapReduce. Я хочу отбросить все строки в исходном файле, у которого есть хотя бы 1 верхняя буква в статусе. Другими словами, я только забочусь о строках, которые имеют все строчные буквы в статусе.

Я новичок в программировании MapReduce и нуждается в некоторой помощи. Ниже то, что я пришел с до сих пор

Мой mapper.py является:

import sys 
import re 

for line in sys.stdin: 
    line = line.strip() 
    portions = re.split(r'\t+', line) 
    status = portions[-1] #pop out last item (status info) from portions list 
    #Now I want to emit key as status and value as the portions list 
    print '%s\t%s' % (status, portions) #obviously, I don't think it's correct, I got stuck at this part 

мой reducer.py является:

import sys 

#I'm assuming that I have read in status and portions. 
#In my understanding, the number of output files depend on the number of reducers. But what I want is to discard all rows that has at least 1 upper case letter in status bar. 
#The file output should be a single file with rows that have all lower case letters in status bar. 

Я думаю, что проверка, если статус имеет, по меньшей мере, 1 верхняя дело не сложно, но я застрял в том, как отбрасывать строки, которые мне не интересны, и объединить все выходные файлы в один файл с тем же форматом, что и в исходном текстовом файле.

Любая помощь, направленная мне на правильный путь, очень ценится. Заранее спасибо!

Answer 1

Вы могли бы сделать это без редуктора на всех,

что-то вроде этого в вашем картографа:

import sys 
import re 

for line in sys.stdin: 
    line = line.strip() 
    portions = re.split(r'\t+', line) 
    status = portions[-1] 
    if status.islower(): 
     whatever_you_want_to_write = status + ',' + portions #whatever 
     sys.stdout.write(whatever_you_want_to_write) 

Смотри документацию Hadoop Streaming для получения подробной информации о чтении/записи в HDFS.

Нечто подобное, например:

$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar \ 
    -input myInputDirs \ 
    -output myOutputDir \ 
    -mapper myPythonScript.py \ 
    -jobconf mapred.reduce.tasks=0 \ 
    -file myPythonScript.py 

Обратите внимание, как вы можете указать -jobconf mapred.reduce.tasks=0 сказать Hadoop, что вы не будете нуждаться в свертке шага.