Использование Hadoop в Python для обработки большого файла CSV - PullRequest
Новая
       26

Использование Hadoop в Python для обработки большого файла CSV

3 голосов
/ 13 февраля 2012

У меня есть огромный CSV-файл, который я хотел бы обработать с помощью Hadoop MapReduce в Amazon EMR (python).

Файл содержит 7 полей, однако я только смотрю на поля date и amount . 

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

Во-первых, мой картограф.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)

Для редуктора я буду использовать потоковую команду: aggregate.

Вопрос:

  1. Прав ли мой код?Я запустил его в Amazon EMR, но получил пустой вывод.

  2. Таким образом, мой конечный результат должен быть: экспресс, XXX и неэкспресс, YYY.Можно ли сделать операцию деления перед возвратом результата?Просто результат XXX / YYY.Где я должен положить этот код?Редуктор ??

  3. Кроме того, это огромный CSV-файл, поэтому отображение разбито на несколько разделов?Или мне нужно явно вызвать FileSplit?Если да, то как мне это сделать?

1 Ответ

3 голосов
/ 15 февраля 2012

Отвечая на мой вопрос здесь!

  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. У меня пока нет ответа.

...