Я хочу решить проблему подсчета слов и хочу получить результаты в обратном порядке в зависимости от частоты появления в файле.
Ниже приведены четыре файла (2 преобразователя и 2 преобразователя, какодна задача Map Reduce не может решить эту проблему) Я написал для этой цели:
1) mapper1.py
import sys
import re
reload(sys)
sys.setdefaultencoding('utf-8') # required to convert to unicode
for line in sys.stdin:
try:
article_id, text = unicode(line.strip()).split('\t', 1)
except ValueError as e:
continue
words = re.split("\W*\s+\W*", text, flags=re.UNICODE)
for word in words:
print "%s\t%d" % (word.lower(), 1)
2) reducer1.py
import sys
current_key = None
word_sum = 0
for line in sys.stdin:
try:
key, count = line.strip().split('\t', 1)
count = int(count)
except ValueError as e:
continue
if current_key != key:
if current_key:
print "%s\t%d" % (current_key, word_sum)
word_sum = 0
current_key = key
word_sum += count
if current_key:
print "%s\t%d" % (current_key, word_sum)
3) mapper2.py
import sys
import re
reload(sys)
sys.setdefaultencoding('utf-8') # required to convert to unicode
for line in sys.stdin:
try:
word, count = line.strip().split('\t', 1)
count = int(count)
except ValueError as e:
continue
print "%s\t%d" % (word, count)
4) reducer2.py
import sys
for line in sys.stdin:
try:
word, count = line.strip().split('\t', 1)
count = int(count)
except ValueError as e:
continue
print "%s\t%d" % (word, count)
Ниже приведены две команды yarn, выполняемые мной в среде bash
OUT_DIR="wordcount_result_1"
NUM_REDUCERS=8
hdfs dfs -rm -r -skipTrash ${OUT_DIR} > /dev/null
yarn jar /opt/cloudera/parcels/CDH/lib/hadoop-mapreduce/hadoop-streaming.jar \
-D mapred.jab.name="Streaming wordCount" \
-D mapreduce.job.reduces=${NUM_REDUCERS} \
-files mapper1.py,reducer1.py \
-mapper "python mapper1.py" \
-combiner "python reducer1.py" \
-reducer "python reducer1.py" \
-input /test/articles-part-short \
-output ${OUT_DIR} > /dev/null
OUT_DIR_2="wordcount_result_2"
NUM_REDUCERS=1
hdfs dfs -rm -r -skipTrash ${OUT_DIR_2} > /dev/null
yarn jar /opt/cloudera/parcels/CDH/lib/hadoop-mapreduce/hadoop-streaming.jar \
-D mapred.jab.name="Streaming wordCount Rating" \
-D mapreduce.job.output.key.comparator.class=org.apache.hadoop.mapreduce.lib.partition.KeyFieldBasedComparator \
-D map.output.key.field.separator=\t \
-D mapreduce.partition.keycomparator.options=-k2,2nr \
-D mapreduce.job.reduces=${NUM_REDUCERS} \
-files mapper2.py,reducer2.py \
-mapper "python mapper2.py" \
-reducer "python reducer2.py" \
-input ${OUT_DIR} \
-output ${OUT_DIR_2} > /dev/null
hdfs dfs -cat ${OUT_DIR_2}/part-00000 | head
Это не дает мне правильный ответ.Может кто-нибудь объяснить, где это пошло не так?
С другой стороны,
в mapper2.py, если я печатаю следующим образом,
print "%d\t%s" % (count, word)
и в reducer2.py, если я прочитаю следующим образом,
count, word = line.strip().split('\t', 1)
и измените параметр команды 2-й пряжи на
-D mapreduce.partition.keycomparator.options=-k1,1nr
дает правильный ответ.
Почему в обоих случаях он ведет себя по-разному?
Может кто-нибудь помочь мне разобраться в параметрах компаратора в Hadoop MapReduce?