hadoop表拼接简单介绍

在使用hadoop进行计算时，经常遇到的一种场景就是拼接两个表，然后进行简单的处理，这种场景可以通过以下这种较为通用的逻辑来进行处理：

#-*- coding: gbk -*-

import sys, string, gc

def mapper() :
    """
    在query后加入A或B字段
    """
    for line in sys.stdin:
        line = line.rstrip()
        if line == '' :
            continue
        fields = line.split('\t')
        fields_len = len(fields)
        if fields_len == 4 :
            print '%s\t%s' % (fields[0], 'A')
        if fields_len > 50 :
            print '%s\t%s\t%s' % (fields[11], 'B', fields[2])


def reducer() :
    """
    合并
    """
    key = ''
    value = 0
    for line in sys.stdin :
        line = line.rstrip()
        if line == '' :
            continue
        fields = line.split('\t')
        if len(fields) < 2 :
            continue
        if fields[1] == 'A' :
            key = fields[0]
        if fields[0] == key and fields[1] == 'B' :
            value += string.atof(fields[2])

    print 'final_sum : %.2f' % (value)

if __name__ == '__main__' :
    if(sys.argv[1] == 'map'):
        mapper()
    elif (sys.argv[1] == 'reduce'):
        reducer()

mapper阶段：通过数据若干属性的不同进行区分并且标记；
                       在本例中是通过列数的不同，从而进行区分，并在第二列标记A，B，第一列的相同属性作为key，并根据需求输出reducer阶段所需的少数几个字段；
reducer阶段：利用hadoop以第一列作为key，对mapper结果进行排序后的顺序结果，进行简单的处理；
                       在本例中是通过先将A数据的第一列赋值给key，然后在B数据中第一列等于key的数据中，累加其第二列，最后输出所有累加值之和；
当然，该hadoop任务的执行需要配置特定的参数才行，下篇将对hadoop参数做简单的介绍，谢谢！