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

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

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

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

Установка Spark в локальном режиме практически одинакова для основных операционных систем и сводится к действиям:
1. Скачиваем Spark ^[6] (данный пример работает для версии 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 ^[7].

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

Осталось «подготовить» наш файл для SQL-запросов (в версии Spark 1.6 для некоторых типов файлов можно напрямую делать SQL-запросы без создания таблицы ^[9]). Т. е. создадим DataFrame (у DataFrame тоже куча полезных функций ^[10]) и из него — таблицу для SQL-запросов:
1. Загружаем нужные библиотеки

>>> from pyspark.sql import SQLContext, Row
>>> sqlContext = SQLContext(sc)

2. Заводим переменную text как исходный файл для обработки и смотрим что в первой строчке:

>>> text = sc.textFile('путь к файлу')
>>> text.first()
u'2015-09-01tu'день знаний'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 ^[10]):

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

Автор: kpimaker

Источник ^[11]