在Python中使用Hadoop处理大csv文件

Question

我有一个很大的CSV文件，想用Hadoop MapReduce在亚马逊EMR上处理（用Python）。

这个文件有7个字段，但我只关注日期和数量这两个字段。

 "date" "receiptId" "productId" "quantity"  "price" "posId" "cashierId"

首先，这是我的mapper.py文件。

import sys

def main(argv):
    line = sys.stdin.readline()
    try:
        while line:
            list = line.split('\t')

            #If date meets criteria, add quantity to express key
                if int(list[0][11:13])>=17 and int(list[0][11:13])<=19:
                    print '%s\t%s' % ("Express", int(list[3]))
            #Else, add quantity to non-express key
                else:
                    print '%s\t%s' % ("Non-express", int(list[3]))

            line =  sys.stdin.readline()
except "end of file":
        return None
if __name__ == "__main__":
        main(sys.argv)

对于reducer，我会使用流式命令：aggregate。

问题：

1. 我的代码对吗？我在亚马逊EMR上运行了它，但输出是空的。

2. 我的最终结果应该是：快递，XXX和非快递，YYY。我能在返回结果之前做一个除法操作吗？就是XXX/YYY的结果。我应该把这段代码放在哪里？在reducer里吗？

3. 另外，这个CSV文件很大，映射会把它分成几个部分吗？还是我需要明确调用FileSplit？如果需要，我该怎么做？

Answer 1

我在这里回答我自己的问题！

1. 代码是错的。如果你使用聚合库来进行数据的归约，你的输出就不会是通常的键值对格式。它需要一个“前缀”。

   if int(list[0][11:13])>=17 and int(list[0][11:13])<=19:
       #This is the correct way of printing for aggregate library
       #Print all as a string.
       print  "LongValueSum:" + "Express" + "\t" + list[3]
   

   其他可用的“前缀”有：DoubleValueSum（双精度值求和）、LongValueMax（长整型值最大值）、LongValueMin（长整型值最小值）、StringValueMax（字符串最大值）、StringValueMin（字符串最小值）、UniqValueCount（唯一值计数）、ValueHistogram（值直方图）。想了解更多信息，可以查看这里 http://hadoop.apache.org/common/docs/r0.15.2/api/org/apache/hadoop/mapred/lib/aggregate/package-summary.html。

2. 是的，如果你想做的不仅仅是基本的求和、最小值、最大值或计数，你需要自己写一个归约器。

3. 我还没有答案。