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

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

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

Установка 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 (пример: http://ift.tt/1iUOvRS). При успешном запуске получаем:

Осталось «подготовить» наш файл для 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, с которым можно совершать много полезных операций обработки (примеры http://ift.tt/1OOEfZZ):

>>> 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')

This entry passed through the Full-Text RSS service - if this is your content and you're reading it on someone else's site, please read the FAQ at http://ift.tt/jcXqJW.