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 (пример: spark.apache.org/docs/latest/quick-start.html). При успешном запуске получаем: 
Осталось «подготовить» наш файл для 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')
