---

Декабрь 2015

ТОМ 30, НОМЕР 13

Тесты - Введение в Spark для .NET-разработчиков

Джеймс Маккафри | Декабрь 2015

Spark — это вычислительная инфраструктура с открытым исходным кодом, предназначенная для работы с так называемыми большими данными (Big Data); она становится все популярнее, особенно в сценариях с машинным обучением. В этой статье я опишу, как установить Spark на компьютер под управлением операционной системы Windows, и поясню базовую функциональность с точки зрения .NET-разработчика.

Лучший способ понять, о чем эта статья, взглянуть на демонстрационный интерактивный сеанс на рис. 1. Из командной оболочки Windows, работающей в административном режиме, я запустил среду Spark командой spark-shell.

Рис. 1. Spark в действии

Команда spark-shell генерирует интерпретатор Scala, который выполняется в оболочке и в свою очередь выводит приглашение к вводу в Scala (scala>). Scala — это скриптовый язык, основанный на Java. Существуют и другие способы взаимодействия со Spark, но использование интерпретатора Scala — наиболее распространенный подход отчасти потому, что инфраструктура Spark написана по большей части на Scala. Вы также можете взаимодействовать со Spark, используя команды языка Python или создавая программы на Java.

Обратите внимание на несколько предупреждений на рис. 1. Эти сообщения очень часто появляются при запуске Spark, так как Spark имеет множество необязательных компонентов, в отсутствие которых генерируются предупреждения. В принципе, эти сообщения можно игнорировать в простых сценариях.

Первая команда, введенная в демонстрационном сеансе:

scala> val f = sc.textFile("README.md")

В вольной интерпретации это означает: «сохранить содержимое файла README.md в неизменяемый RDD-объект f». Объекты Scala могут объявляться как val или var. Объекты, объявленные как val, являются неизменяемыми.

В интерпретатор Scala встроен объект контекста Spark с именем sc, который используется для доступа к функциональности Spark. Функция textFile загружает содержимое текстового файла в структуру данных Spark, называемую отказоустойчивым распределенных набором данных (resilient distributed dataset, RDD). RDD являются основной программной абстракцией, применяемой в Spark. Вы можете считать RDD чем-то схожим с .NET-набором, хранящимся в оперативной памяти и распределенным по нескольким машинам.

Текстовый файл README.md (расширение .md обозначает «markdown document» [документ разметки]) находится в корневом каталоге C:\spark_1_4_1. Если ваш целевой файл расположен где-то еще, вы можете указать полный путь к нему, например C:\\Data\\ReadMeToo.txt.

Вторая команда в демонстрационном сеансе:

scala> val ff = f.filter(line => line.contains("Spark"))

Это означает: «Сохранить в неизменяемый RDD-объект с именем ff только те строки из объекта f, в которых есть слово 'Spark'». Функция filter принимает так называемое замыкание (closure). Замыкание можно рассматривать как нечто похожее на анонимную функцию. Здесь замыкание принимает строковый входной параметр line и возвращает true, если line содержит «Spark», или false в ином случае.

Поскольку line — просто имя параметра, я мог бы использовать в замыкании любое другое имя, скажем:

ln => ln.contains("Spark")

Spark чувствительна к регистру букв, поэтому следующее привело бы к генерации ошибки:

ln => ln.Contains("Spark")

Scala обладает некоторыми характеристиками языка функционального программирования и позволяет составлять несколько команд. Например, первые две команды можно было бы объединить в одну:

val ff = sc.textFile("README.md").filter(line => lne.contains("Spark"))

Последние три команды в демонстрационном сеансе выглядят так:

scala> val ct = ff.count()
scala> println(ct)
scala> :q

Функция count возвращает количество элементов в RDD — в данном случае число строк в файле README.md, содержащий слово «Spark». Таких строк — 19. Чтобы выйти из сеанса Spark Scala, введите команду :q.

Установка Spark на машину с Windows

Процесс установки Spark на машину с Windows состоит из четырех основных этапов. Во-первых, вы устанавливаете Java Development Kit (JDK) и Java Runtime Environment (JRE). Во-вторых, вы устанавливаете язык Scala. В-третьих, вы устанавливаете инфраструктуру Spark. И в-четвертых, конфигурируются системные переменные на хост-машине.

Дистрибутив Spark предоставляется в сжатом формате .tar, поэтому вам понадобится утилита для извлечения файлов Spark. Советую установить программу 7-Zip с открытым исходным кодом.

Хотя Spark и ее компоненты не имеют формальной поддержки на широком спектре версий Windows, я успешно устанавливал Spark на компьютеры под управлением Windows 7, 8, 10 и Server 2008, 2012. Демонстрация на рис. 1 выполнялась на компьютере с Windows 8.1.

JDK устанавливается запуском самораспаковывающегося исполняемого файла, который вы найдете поиском в Интернете. Я использовал версию jdk-8u60-windows-x64.exe.

При установке 64-разрядной версии JDK каталогом установки по умолчанию является C:\Program Files\Java\jdkx.x.x_xx\, как показано на рис. 2. Рекомендую не изменять этот каталог по умолчанию.

Рис. 2. Каталог установки JDK по умолчанию

При установке JDK устанавливается и связанная с ним JRE. По окончании установки родительский каталог Java по умолчанию будет содержать каталог JDK и связанный каталог JRE, как показано на рис. 3.

.png "Java JDK и JRE, установленные в C:\Program Files\Java\")
Рис. 3. Java JDK и JRE, установленные в C:\Program Files\Java\

Заметьте, что у вас в системе скорее всего также появится каталог Java с одним или более каталогами 32-разрядной JRE в C:\Program Files (x86). Это нормально, когда в системе установлены 32- и 64-разрядные версии JRE, но я советую пользоваться только 64-разрядной версией Java JDK.

Установка Scala

Следующий этап — установка языка Scala, но сначала вы должны отправиться на сайт для скачивания Spark (описывается в следующем разделе этой статьи) и определить, какую версию Scala нужно установить. Версия Scala должна быть совместимой с версией Spark, которую вы будете устанавливать на следующем этапе.

К сожалению, информация о совместимости версий Scala и Spark весьма скудная. Когда я устанавливал компоненты Spark (некоторое время назад), текущей версией Spark была 1.5.0, но я сумел найти никакой информации о том, какая версия Scala совместима с этой версией Spark. Поэтому я предпочел предыдущую версию Spark (1.4.1), найдя кое-какую информацию на форумах для разработчиков, где сообщалось, что Scala версии 2.10.4 скорее всего совместима со Spark версии 1.4.1.

Установить Scala легко. Для этого достаточно запустить файл установщика .msi.

Процессом руководит мастер установки Scala. Любопытно, что каталог установки по умолчанию для Scala находится в каталоге для 32-разрядных программ — C:\Program Files (x86)\, а не rather в каталоге для 64-разрядных — C:\Program Files\ (рис. 4).

.png "Scala устанавливается в C:\Program Files (x86)\scala\")
Рис. 4. Scala устанавливается в C:\Program Files (x86)\scala\

Если вы намерены взаимодействовать со Spark, создавая программы на Java, а не используя команды Scala, вам понадобится дополнительная утилита — Scala Simple Build Tool (SBT). Взаимодействие со Spark через скомпилированные Java-программы гораздо труднее, чем через интерактивную среду Scala.

Установка Spark

Следующий шаг — установка инфраструктуры Spark. Но сначала убедитесь, что у вас есть вспомогательная программа наподобие 7-Zip, способная распаковывать файлы формата .tar. Процесс установки Spark осуществляется вручную: вы скачиваете сжатую папку на локальную машину, извлекаете сжатые файлы и копируете их в корневой каталог. А значит, если вы пожелаете удалить Spark, то просто удалите файлы Spark.

Сайт Spark — spark.apache.org. Страница скачивания позволяет выбрать версию и тип пакета. Spark — это вычислительная инфраструктура, требующая наличия распределенной файловой системы (distributed file system, DFS). Пока что самая распространенная DFS, используемая с инфраструктурой Spark, — распределенная файловая система Hadoop (HDFS). Для тестирования и экспериментов, таких как демонстрационный сеанс на рис. 1, можно установить Spark в систему, не имеющую DFS. В этом сценарии Spark будет использовать локальную файловую систему.

Если раньше вы не распаковывали файлы .tar, этот процесс может показаться вам слегка запутанным, поскольку, как правило, извлекать файлы приходится дважды. Сначала скачайте файл .tar (у меня такой файл называется spark-1.4.1-bin-hadoop2.6.tar) в любой временный каталог (я использовал C:\Temp). Затем щелкните правой кнопкой мыши файл .tar, выберите Extract files из контекстного меню и извлеките его содержимое в новый каталог внутри временного.

На первом проходе распаковки создается новый сжатый файл безо всякого расширения (в моем случае — spark-1.4.1-bin-hadoop2.6). Затем щелкните правой кнопкой мыши этот новый файл, снова выберите Extract files из контекстного меню и извлеките его содержимое в другой каталог. После второго прохода создаются файлы инфраструктуры Spark.

Создайте каталог для этих файлов. По распространенному соглашению, нужно создать каталог C:\spark_x_x_x, где x — значения, указывающие версию. Используя это соглашение, я создал каталог C:\spark_1_4_1 и скопировал в него извлеченные файлы (рис. 5).

.png "Ручное копирование извлеченных файлов Spark в C:\spark_x_x_x\")
Рис. 5. Ручное копирование извлеченных файлов Spark в C:\spark_x_x_x\

Настройка вашего компьютера

После установки Java, Scala и Spark остается сконфигурировать хост-машину. Этот процесс включает скачивание специального вспомогательного файла для Windows, настройку трех системных переменных окружения, определенных пользователем, задание переменной окружения Path и (не обязательно) модификацию конфигурационного файла Spark.

Запуск Spark в Windows требует, чтобы этот специальный вспомогательный файл, winutils.exe, находился в локальном каталоге C:\hadoop. Вы можете найти этот файл в нескольких местах поиском по Интернету. Я создал каталог C:\hadoop, а затем нашел копию winutils.exe по ссылке public-repo-1.hortonworks.com/hdp-win-alpha/winutils.exe и скачал этот файл в свой каталог.

Далее создаем и настраиваем системные переменные окружения, определяемые пользователем, и модифицируем переменную окружения Path. Перейдите в Control Panel | System | Advanced System Settings | Advanced | Environment Variables. В разделе User Variables создайте три новые переменные с указанными ниже именами и значениями:

JAVA_HOME     C:\Program Files\Java\jdk1.8.0_60
SCALA_HOME    C:\Program Files (x86)\scala
HADOOP_HOME   C:\hadoop

Затем в разделе System Variables измените переменную Path, добавив местонахождение двоичных файлов Spark — C:\spark_1_4_1\bin. Будьте осторожны, чтобы не потерять никакие другие значения в переменной Path. Заметьте, что процесс установки Scala уже добавил местонахождение двоичных файлов Scala за вас (рис. 6).

Рис. 6. Настройка вашей системы

После настройки системных переменных советую модифицировать конфигурационный файл Spark. Перейдите в корневой каталог C:\spark_1_4_1\config и сделайте копию файла log4j.properties.template. Переименуйте эту копию, удалив расширение .template. Измените первый элемент конфигурации с log4j.rootCategory=INFO на log4j.rootCategory=WARN.

Идея в том, что по умолчанию Spark выдает все виды информационных сообщений. Смена уровня протоколирования с INFO на WARN значительно уменьшает количество сообщений, не так сильно мешая взаимодействию со Spark.

«Hello World» в Spark

Пример Hello World в распределенных вычислениях — это подсчет числа разных слов в источнике данных. На рис. 7 показан пример подсчета слов с использованием Spark.

Рис. 7. Пример подсчета слов с использованием Spark

Оболочку Scala иногда называют оболочкой цикла чтения, оценки и вывода (read, evaluate, print loop, REPL). Вы можете очистить Scala REPL, нажав комбинацию клавиш CTRL+L. Первая команда на рис. 7 загружает содержимое файла README.md в RDD с именем f, как пояснялось ранее. На практике источник данных может оказаться огромным файлом, распределенным по сотням машин, или распределенной базой данных вроде Cassandra.

Следующая команда выглядит так:

scala> val fm = f.flatMap(line => line.split(" "))

Вызов функции flatMap разбивает каждую строку в RDD-объекте f по пробельным символам, поэтому конечный RDD-объект fm будет хранить набор всех слов в файле. С точки зрения разработчика, fm можно рассматривать как нечто подобное .NET-набору List<string>.

Следующая команда:

scala> val m = fm.map(word => (word, 1))

Функция map создает RDD-объект, который хранит пары элементов, где каждая пара состоит из слова и целочисленного значения 1. Вы можете увидеть это более наглядно, если введете команду m.take(5). Тогда у вас появится первые пять слов из файла README.md и значение 1 рядом с каждым словом. С Точки зрения разработчика, m примерно соответствует набору List<Pair>, где каждый объект Pair состоит из строки и целого значения. Строка (слово из README.md) является ключом, а целочисленный тип — значением, но в Spark в отличие от многих пар «ключ-значение» в Microsoft .NET Framework допускается наличие дубликатов для пар «ключ-значение». RDD-объекты, содержащие пары «ключ-значение», иногда называют парными, чтобы отличать их от обычных RDD.

Очередная команда:

scala> val cts = m.reduceByKey((a,b) => a + b)

Функция reduceByKey объединяет элементы в объекте m, суммируя целые значения, сопоставленные с одинаковыми ключами. Выполнив команду cts.take(10), вы увидите 10 слов из README.md, за которыми следует число раз, которое каждое слово встречается в файле. Кроме того, вы, вероятно, заметите, что слова в объекте cts не обязательно следуют в каком-то определенном порядке.

Функция reduceByKey принимает замыкание. Поэтому можно использовать альтернативную сокращенную нотацию Scala:

scala> val cts = m.reduceByKey(_ + _)

Знак подчеркивания является символом подстановки параметра, поэтому синтаксис можно интерпретировать как «суммировать любые полученные два значения».

Заметьте, что этот пример с подсчетом слов использует функцию map, за которой вызывается функция reduceByKey. Вот пример парадигмы MapReduce:

scala> val sorted =
     cts.sortBy(item => item._2, false)

Эта команда сортирует элемент (item) в RDD-объекте cts на основе второго значения (целочисленного счетчика) элементов. Аргумент false означает, что сортировать нужно в порядке убывания, т. е. от наибольшего значения к наименьшему. Форма сокращенного синтаксиса Scala для команды sort выглядит так:

scala> val sorted = cts.sortBy(_._2, false)

Поскольку Scala обладает многими характеристиками языка функционального программирования и использует множество символов вместо ключевых слов, на Scala можно написать крайне непонятный код.

Последняя команда в примере Hello World отображает результаты:

scala> sorted.take(5).foreach(println)

Это означает: «Извлечь первые пять объектов в RDD-объекте с именем sorted, перебрать этот набор, применяя функцию println к каждому элементу». Вот результаты:

(,66)
(the,21)
(Spark,14)
(to,14)
(for,11)

Таким образом, мы имеем 66 пустых/null слов в README.md, 21 слово «the», 14 слов «Spark» и т. д.

Заключение

Информация, представленная в этой статье, должна помочь вам освоить азы и приступить к экспериментам со Spark на машине с Windows. Spark — относительно новая технология, созданная в Калифорнийском университете в Беркли (2009 год), но интерес к Spark резко возрос в последние несколько месяцев, по крайней мере среди моих коллег.

На конкурсе инфраструктур обработки больших данных в 2014 году Spark установила новый рекорд в производительности, легко побив предыдущий рекорд, установленный системой Hadoop год назад. Из-за своей исключительной производительности Spark особенно хорошо подходит для использования с системами машинного обучения. Spark поддерживает библиотеку алгоритмов машинного обучения MLib с открытым исходным кодом.

---

Джеймс Маккафри (Dr. James McCaffrey) — работает на Microsoft Research в Редмонде (штат Вашингтон). Принимал участие в создании нескольких продуктов Microsoft, в том числе Internet Explorer и Bing. С ним можно связаться по адресу jammc@microsoft.com.

Выражаю благодарность за рецензирование статьи экспертам Microsoft Газу Икбалу (Gas Iqbal) и Умешу Мадану (Umesh Madan).