在Python中使用Hadoop来处理一个大的csv文件
问题描述:
我有一个巨大的CSV文件,我想在Amazon EMR(python)上使用Hadoop MapReduce进行处理。在Python中使用Hadoop来处理一个大的csv文件
该文件有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)
对于减速,我将使用流命令:集料。
问:
是我的代码吗?我在Amazon EMR中运行它,但是我得到了一个空输出。
所以我最终的结果应该是:表达,XXX和非表达,YYY。在返回结果之前,我可以让它做分割操作吗?只是XXX/YYY的结果。我应该在哪里放这个代码?减速器??
此外,这是一个巨大的CSV文件,所以将映射分解成几个分区?或者我需要显式调用FileSplit?如果是这样,我该怎么做?
答
在这里回答我自己的问题!
-
The code is wrong。如果您使用聚合库进行缩减,则您的输出不会遵循常用的键值对。它需要一个“前缀”。
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。
是的,如果您想要做的不仅仅是基本总和,最小值,最大值或者计数,您需要编写自己的reducer。
我还没有答案。
为什么不使用python内置的csv解析器? – 2012-02-14 23:18:15