MR 全局排序——单reducer

约定不等于承诺〃 2023-06-12 09:54 12阅读 0赞

利用MapReduce框架完成上述全局排序，将怎样来处理呢？我们知道，map的输出结果是键值对的形式，框架先将一行行的键值数据分区，同一个分区的数据聚集在一起，每个分区内的数据按照key排序，然后每个分区内的数据会被分发到对应的reduce处理，最后reduce输出结果，这个过程叫shuffle。依据此，设定一个reduce task，即强制把map输出的结果都分到一个分区中，也就交给了一个reduce处理，再以第一个字段为key，交由MapReduce去排序。

# 方式一：通过添加一个变量base\_count，利用字典序进行全局排序 #

## map\_sort.py ##

key加上base\_count，使得结果key长度一致，可以用字典序进行排序。

#! /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)
    
                    print "%s\t%s" %(new_key,val)
    except Exception:
        print "map error"

## red\_sort.py ##

在reduce中，key要减去base\_count，以此还原成原先的key。

#! /usr/bin/python
    
    import sys
    
    base_count = 10000
    try:
            for line in sys.stdin:
                    key, val = line.strip().split('\t')
                    print str(int(key)-base_count) + '\t' + val
    except Exception:
        print "reduce error"

map和reduce函数只是将数据作了分割然后输出的处理，排序的功能交由MapReduce框架自动处理

## run.sh ##

这里设置reduce task的数目为1，即所有数据都会进入到一个partition中，这时利用框架自带的排序功能（同一partition内的数据按照key**升序**），看看能否对数据进行全局排序。hadoop streaming会默认把map输出结果中每一行第一个分隔符（默认为"\\t"）前的字段作为key，后面的作为value，并且若不指定以key中哪个字段作partition，则默认以整个key字段作partition，但这里指定reduce task为1，强制只有1一个partition了。

结果：默认是升序

![20191112162227803.png][]

![20191112151721741.png][]

# 方式二：通过配置参数 #

## map\_sort.py ##

#! /usr/bin/python
    
    import sys
    
    try:
            for line in sys.stdin:
                    ss = line.strip().split('\t')
                    key = ss[0]
                    val = ss[1]
    
                    print "%s\t%s" %(key,val)
    except Exception:
        print "map error"

## red\_sort.py ##

#! /usr/bin/python
    
    import sys
    
    try:
            for line in sys.stdin:
                    print line.strip()
    
    except Exception:
        print "reduce error"

## run.sh ##

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_reduceByOne2/a.txt"
    INPUT_FILE_PATH_B="/Test/mr_allsort_reduceByOne2/b.txt"
    
    OUTPUT_PATH="/Test/mr_allsort_reduceByOne2/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 \
            -mapper "python map_sort.py" \
            -reducer "python red_sort.py" \
            -file ./map_sort.py \
            -file ./red_sort.py \        
            # 利用该配置完成二次排序
            -partitioner org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner \
            # 利用该配置完成key排序
            -jobconf mapred.output.key.comparator.class=org.apache.hadoop.mapred.lib.KeyFieldBasedComparator \
            # 设置map分隔符的位置，该位置前的为key，之后的为value
            -jobconf stream.num.map.output.key.fields=1 \
            #选择哪一部分做partition
            -jobconf mapred.text.key.partitioner.options="-k1,1" \
            #设置key中需要比较的字段或字节范围
            -jobconf mapred.text.key.comparator.options="-k1,1n" \
            # reduce任务的个数
            -jobconf mapred.reduce.tasks=1

[20191112162227803.png]: https://img-blog.csdnimg.cn/20191112162227803.png
[20191112151721741.png]: https://img-blog.csdnimg.cn/20191112151721741.png