MR 全域性排序 多reducer

在資料量大的情況下，使用單reducer進行全域性排序的方式明顯效率較低，可次採用多reducer的方式。

在map中進行分桶，分桶方式自定。

#! /usr/bin/python

import sys
base_count = 10000
try:
for line in sys.stdin:
ss = line.strip().split('\t')
key = ss[0]
val = ss[1]
new_key = base_count + int(key)
partition_id = 1
if new_key <= (10000+10100) / 2:
partition_id = 0
print "%s\t%s\t%s" %(partition_id, new_key, val)
except exception:
print "map error"

#! /usr/bin/python

import sys
try:
for line in sys.stdin:
partition_id, key, val = line.strip().split('\t')
print '\t'.join([key,val])
except exception:
print "reduce error"

set -e -x

hadoop_cmd="/usr/local/src/hadoop-2.6.5/bin/hadoop"
stream_jar_path="/usr/local/src/hadoop-2.6.5/share/hadoop/tools/lib/hadoop-streaming-2.6.5.jar"
input_file_path_a="/test/mr_allsort_reducebymany/a.txt"
input_file_path_b="/test/mr_allsort_reducebymany/b.txt"
output_path="/test/mr_allsort_reducebymany/result"
$hadoop_cmd fs -rm -r -skiptrash $output_path
$hadoop_cmd jar $stream_jar_path \
-input $input_file_path_a,$input_file_path_b \
-output $output_path \
-reducer "python red_sort.py" \
-file ./map_sort.py \
-file ./red_sort.py \
# 定義兩個reduce任務
-jobconf mapred.reduce.tasks=2 \
# map分隔符位於第二個欄位後，前兩個字段作為key，後面的作為value
-jobconf stream.num.map.output.key.fields=2 \
# 按分隔符分割後，第乙個字段作為分割槽字段
-jobconf num.key.fields.for.partition=1 \
-partitioner org.apache.hadoop.mapred.lib.keyfieldbasedpartitioner

org.apache.hadoop.mapred.lib.keyfieldbasedpartitioner 如果要設定key中用於partition的字段，而不是把整個key都用來做partition，就用此配置項

結果：由於定義了兩個reduce任務，所以生成兩個結果檔案。

MR 二次排序

二次排序 eg 對左側序列進行排序,需要先對字母排序,然後在對數字進行排序.得到右側的序列.a 2 a 2 c 4 a 4 b 3 b 1 c 1 b 3 a 4 c 1 b 1 c 4public class sortmapreduce extends configured implements ...

hadoop mr 全域性排序

全域性排序 需求 根據使用者每月使用的流量按照使用的流量多少排序 介面 writablecomparable 排序操作在hadoop中屬於預設的行為。預設按照字典順序排序。排序的分類 1 部分排序 2 全排序 3 輔助排序 4 二次排序 封裝類，直接完成排序 public class flowbea...

4 ，mr 八步練習 排序

用 k2 排序 把需要排序的列，封裝到 k2 中，讓 k2 成為乙個封裝型別 注意看 reduce 雖然是空資料，但是也要遍歷，目的是不要重複的資料只顯示一條。a 1 a 9b 3 a 7b 8 b 10 a 5a 9public class pxpojo implements writableco...