Spark local mode: обработка больших файлов на обычном ноутбуке

    image
    Всем привет.
    4 января вышла новая версия Apache Spark 1.6 с bug fix новыми возможностями обработки больших данных. На Хабре написано немало статей по использованию этого инструмента от введения до опыта использования в проектах. Spark работает на большинстве операционных систем и его можно запускать в локальном режиме даже на обычном ноутбуке. Используя простоту настройки Spark в этом случае грех не воспользоваться основными функциям. В этой статье мы посмотрим как на ноутбуке быстро настроить обработку большого файла (больше оперативной памяти компьютера) с помощью обычных SQL-запросов. Это позволит делать запросы даже неподготовленному пользователю. Дополнительное подключение iPython (Jupyter) notebook позволит составлять полноценные отчеты. В статье разобран простой пример обработки файла, другие примеры на Python есть тут.

    Входные данные: файл(-ы) по несколько GB с упорядоченными данными, ноутбук со свободной RAM < 1GB. Необходимо получать различные аналитические данные с помощью SQL- или подобных несложных запросов к файлам. Разберем пример, когда в файлах лежит статистика поисковых запросов за месяц (данные на скриншотах показаны для примера и не соответствуют действительности):
    image

    Необходимо получить распределение количества слов в поисковом запросе для запросов определенной тематики. Например, содержащих слово «недвижимость». Т. е. в данном примере просто фильтруем поисковые запросы, считаем количество слов в каждом запросе, группируем по количеству слов и строим распределение:
    image

    Установка Spark в локальном режиме практически одинакова для основных операционных систем и сводится к действиям:
    1. Скачиваем Spark (данный пример работает для версии 1.6) и разархивируем в любую папку.

    2. Установка Java (если нет)
    — для Windows и MAC скачиваем и устанавливаем 7 версию с java.com
    — для Linux: $ sudo apt-get update и $ sudo apt-get install openjdk-7-jdk + может понадобиться в .bashrc добавить адрес установки JAVA: JAVA_HOME="/usr/lib/jvm/java-7-openjdk-i386"
    Если нет Python, то можно просто установить Anaconda.

    Запускаем pySpark (можно запустить spark-shell для работы в Scala как на нативном языке): заходим в распакованный архив Spark и в папке bin запускаем pyspark (пример: spark.apache.org/docs/latest/quick-start.html). При успешном запуске получаем:
    image

    Осталось «подготовить» наш файл для SQL-запросов (в версии Spark 1.6 для некоторых типов файлов можно напрямую делать SQL-запросы без создания таблицы). Т. е. создадим DataFrame (у DataFrame тоже куча полезных функций) и из него — таблицу для SQL-запросов:
    1. Загружаем нужные библиотеки
    >>> from pyspark.sql import SQLContext, Row
    >>> sqlContext = SQLContext(sc)
    


    2. Заводим переменную text как исходный файл для обработки и смотрим что в первой строчке:
    >>> text = sc.textFile('путь к файлу')
    >>> text.first()
    u'2015-09-01\tu'день знаний'\t101753'
    


    В нашем файле строчки разделены табуляцией. Для корректного разделения по столбцам используем функции Map и Split, используя в качестве разделителя табуляцию: map(lambda l: l.split('\t')). Выберем из результата разбиения нужные столбцы. Для данной задачи нам потребуется знать количество слов в определенном поисковом запросе. Поэтому возьмем только запрос (столбец query) и количество слов в нем (столбец wc): map(lambda l: Row(query=l[1], wc=len(l[1].split(' ')))).

    Можно взять все столбцы таблицы, чтобы в дальнейшем делать произвольные SQL-запросы к ней:
    map(lambda l: Row(date=l[0], query=l[1], stat=l[2], wc=len(l[1].split(' '))))

    Выполним эти действия в одну строчку
    >>> schema = text.map(lambda l: l.split('\t')).map(lambda l: Row(query=l[1], wc=len(l[1].split(' '))))
    


    Осталось перевести schema в DataFrame, с которым можно совершать много полезных операций обработки (примеры spark.apache.org/docs/latest/sql-programming-guide.html#dataframe-operations):
    >>> df = sqlContext.createDataFrame(schema)
    >>> df.show()
    +--------------------+---+
    | query| wc|
    +--------------------+---+
    |день знаний...| 2|
    | сбербанк онлайн| 2|
    |эхо москвы слушать| 3|
    ...
    


    3. Переведем DataFrame в таблицу, чтобы делать SQL-запросы:
     >>> df.registerTempTable('queryTable')
    


    4. Составляем SQL-запрос для всего файла и выгружаем результат в переменную output:
    >>> output = sqlContext.sql('SELECT wc, COUNT(*) FROM queryTable GROUP BY wc').collect()
    


    Для файла в 2GB при свободной RAM в 700MB такой запрос занял 9 минут. Ход обработки процесса можно видеть в строке вида (… из 53):
    INFO TaskSetManager: Finished task 1.0 in stage 8.0 (TID 61) in 11244 ms on localhost (1/53)

    Можем добавить дополнительные ограничения:
    >>> outputRealty = sqlContext.sql('SELECT wc, COUNT(*) FROM queryTable WHERE query like "%недвижимость%" GROUP BY wc').collect()
    


    Осталось нарисовать по этому распределению гистограмму. Например, можно записать результат output в файл 'output.txt' и рисовать распределение просто в Excel:

    >>> with open('output.txt', 'w') as f:
    ...         f.write('wc \t count \n')
    ...         for line in output:
    ...             f.write(str(line[0]) + '\t' + str(line[1]) + '\n')
    
    Метки:
    • +12
    • 14,3k
    • 7
    Поделиться публикацией
    Похожие публикации
    Комментарии 7
    • 0
      А в чем отличие от стандартных юниксовых пайпов (cat file | grep «never gonna give you up»)?
      • +2
        В приведенном варианте ничем, кроме оверхеда. Удобства работы со спарком в консоли на грош.
        Для ad-hoc задач всё же надо использовать другой интерфейс, благо они есть и для скалы и для питона сейчас.
        • 0
          Это просто пример, реально это можно распараллелить на много машин
        • 0
          Народ вон активно пилит Apache Zeppelin, как раз для интерактивной работы a-la ipython notebook. Но оно пока в инкубаторе и неизвестно по стабильности.
          • +2
            В последнее время Hortonworks активно в него донейтит и будет продвигать как Spark Notebook в HDP (инфо из первых рук), т.ч. есть надежда, что его можно будет хотя бы поставить и запустить без плясок с бубнами
            • 0
              Это приятная новость ,) Я возлагаю некоторые надежды на zeppelin после того как посмотрел на graphlab create. Но прицельно пока не щупал.
              • 0
                Я, честно говоря, пока не возлагаю, т.к. не особо понимаю, чем он лучше связки IPython+PySpark+pandas. Для scala есть scala-notebook, который сильно легче и ставится с пол-оборота (чего не скажешь пока про Zeppelin)

          Только полноправные пользователи могут оставлять комментарии. Войдите, пожалуйста.