Обработка данных при помощи Apache Pig

Учитывая популярность платформы Hadoop, не удивительно, что ее экосистема постоянно развивается. Одной из набирающих популярность областей является создание приложений для платформы Hadoop. Несмотря на то, что разработка приложений в соответствии с концепцией Map и Reduce не слишком сложна, она требует наличия определенных навыков в программировании. Apache Pig меняет такое положение вещей, создавая более простую абстракцию процедурного языка над платформой MapReduce, чтобы реализовать SQL-подобный интерфейс для приложений Hadoop. Таким образом, вместо того, чтобы создавать отдельное приложение MapReduce, вы можете написать простой сценарий на Pig Latin, который автоматически распараллеливается и распределяется между узлами кластера.

Пример сценария Pig Latin

Начнем с простого примера Pig. Одним из интересных применений Hadoop является поиск в большом наборе данных записей, удовлетворяющих заданному критерию поиска (в Linux® это делает всем известная команда grep). В листинге 1 продемонстрировано простое решение этой задачи с помощью Pig. Ниже приведены три строки, но непосредственно поиск осуществляется только в одной из них. Первая строка просто считывает тестовый набор данных (log-файл, содержащий сообщения) в массив, представляющий собой набор записей. Затем эти данные фильтруются (каждая запись состоит из одного столбца, идентифицируемого как $0 или поле 1) с помощью регулярного выражения, которое ищет последовательность символов WARN. Наконец, мы сохраняем полученный массив (который теперь содержит только записи, содержащие слово WARN) в новом файле с названием warnings в файловой системе хоста.

messages = LOAD 'messages';
warns = FILTER messages BY $0 MATCHES '.*WARN+.*';
STORE warns INTO 'warnings';

Как видите, в этом простом сценарии реализован простой процесс, который бы потребовал написания значительно большего объема кода, если бы реализовывался непосредственно в традиционной модели MapReduce. По сравнению с традиционным написанием приложений для MapReduce это позволяет существенно упростить изучение Hadoop и быстрее приступить к работе с данными.

Теперь рассмотрим язык Pig и его возможности более подробно.

Основы Pig Latin

Pig Latin - относительно простой язык, выполняющий операторы. Оператор - это процедура, принимающая входные данные (например, массив, состоящий из набора записей) и формирующая на выходе другой массив. Массив - это структура, подобная таблице реляционной базы данных, в которой записи аналогичны строкам таблицы и состоят из полей.

В сценариях Pig Latin часто используется определенный алгоритм, в котором данные считываются из файловой системы, затем над ними выполняется ряд действий (одно или несколько преобразований), после чего конечная структура записывается обратно в файловую систему. Именно этот алгоритм в простейшей форме (с одним преобразованием) был использован в листинге 1.

В Pig имеется богатый набор типов данных - поддерживаются не только высокоуровневые объекты (например, массивы, записи и сопоставления), но также и простые типы данных, например int, long, float, double, chararray и bytearray. К простым типам данных можно применять ряд арифметических операций (например add, subtract, multiply, divide и module), а также условный оператор bincond, который работает так же, как оператор ternary языка C. Как и следует ожидать, в Pig имеется полный набор операторов сравнения, включая сопоставление с шаблонами с помощью регулярных выражений.

Операторы Pig Latin работают с отношениями (и называются реляционными операторами). Из листинга 1 видно, что в нем присутствуют операторы для загрузки и записи данных в файловую систему. Также в нем выполняется фильтрация (FILTER) данных путем последовательного перебора всех строк массива. Эта функциональность широко используется для удаления из массива всех данных, не требующихся для последующих операций. Если же необходимо последовательно перебирать столбцы, а не строки массива, то можно использовать для этого оператор FOREACH, который позволяет вложенным операциям (таким как FILTER и ORDER) преобразовывать данные в процессе итерации.

Оператор ORDER сортирует массив по одному или нескольким полям. Оператор JOIN выполняет внутренние или внешние соединения двух или более массивов в результирующий массив на основе их общих полей. Оператор SPLIT разделяет массив на два или более массива в соответствии с некоторым пользовательским выражением. Наконец, оператор GROUP группирует данные одного или нескольких массивов в соответствии с некоторым выражением. В таблице 1 приведен неполный список реляционных операторов Pig.

Хотя это не исчерпывающий список операторов Pig Latin, в нем представлены наиболее полезные операторы для обработки больших наборов данных. Для Pig написано большое количество онлайновой документации. В оставшейся части этой статьи я предлагаю вам попробовать написать собственные сценарии на Pig Latin, чтобы понять, как эти операторы работают на практике.

Как получить Pig

В предыдущих статьях о Hadoop я предлагал устанавливать и настраивать Hadoop из пакета, однако компания Cloudera упростила этот процесс, создав виртуальную машину ОС Linux с интегрированным в нее Hadoop. В этом случае вам придется загрузить исходный файл большего объема, но зато виртуальная машина уже содержит все настройки конфигурации Hadoop, Apache Hive и Pig. Таким образом, загрузив свободно доступный гипервизор 2 типа (VirtualBox или Kernel-based Virtual Machine [KVM]), вы получите полностью настроенную и готовую к использованию среду Hadoop.

Настройка и запуск Hadoop и Pig

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

Pig можно использовать в одном из двух режимов. Первый из них называется локальным (Local mode); он никак не связан с Hadoop и файловой системой HDFS (Hadoop Distributed File System), а все процессы выполняются в контекстах виртуальной машины Java (JVM) и локальной файловой системы. Второй режим называется режимом Mapreduce (Mapreduce mode); в нем используется файловая систем Hadoop и кластер.

Использование Pig в локальном режиме

Для работы в локальном режиме просто запустите Pig и укажите значение local в опции exectype. В результате вы попадете в командную оболочку Grunt, которая позволяет интерактивно вводить операторы Pig:

$ pig -x local
...
grunt>

После этого вы можете интерактивно вводить код сценария Pig Latin и видеть результаты выполнения каждого оператора. Взгляните еще раз на листинг 1 и попытайтесь воспроизвести этот сценарий (см. листинг 2). Обратите внимание на то, что здесь мы не сохраняем данные в файл, а просто выводим их на экран в виде набора строк. Можно заметить, что в этом отсортированном выводе каждая строка log-файла (удовлетворяющая критерию поиска, заданному с помощью оператора FILTER) сама является массивом ограниченным круглыми скобками ().

grunt> messages = LOAD '/var/log/messages';
grunt> warns = FILTER messages BY $0 MATCHES '.*WARN+.*';
grunt> DUMP warns
...
(Dec 10 03:56:43 localhost NetworkManager: <WARN> nm_generic_enable_loopback(): error ...
(Dec 10 06:10:18 localhost NetworkManager: <WARN> check_one_route(): (eth0) error ...
grunt>

Если бы вы использовали оператор STORE, то данные бы сохранились в директории с указанным именем (а не в обычном файле).

Использование Pig в режиме Mapreduce

Для работы в режиме Mapreduce необходимо, чтобы Hadoop был запущен. Самый простой способ убедиться в этом - это получить список файлов корневой директории файловой системы Hadoop, как показано в листинге 3.

$ hadoop dfs -ls /
Found 3 items
drwxrwxrwx   - hue    supergroup          0 2011-12-08 05:20 /tmp
drwxr-xr-x   - hue    supergroup          0 2011-12-08 05:20 /user
drwxr-xr-x   - mapred supergroup          0 2011-12-08 05:20 /var
$

Если Hadoop запущен и работает, то вы увидите список, состоящий из одного или нескольких файлов, как показано выше. Теперь можно приступать к проверке работы Pig. Начнем с того, что запустим Pig, перейдем в корневую директорию файловой системы hdfs, и посмотрим, увидим ли мы то, что видели в HDFS, не находясь внутри Pig (см. листинг 4).

$ pig
2011-12-10 06:39:44,276 [main] INFO  org.apache.pig.Main - Logging error messages to...
2011-12-10 06:39:44,601 [main] INFO  org.apache.pig.... Connecting to hadoop file system at: hdfs://0.0.0.0:8020
2011-12-10 06:39:44,988 [main] INFO  org.apache.pig.... connecting to map-reduce job tracker at: 0.0.0.0:8021
grunt> cd hdfs:///
grunt> ls
hdfs://0.0.0.0/tmp     <dir>
hdfs://0.0.0.0/user    <dir>
hdfs://0.0.0.0/var     <dir>
grunt>

Пока все в порядке. Мы видим файловую систему Hadoop из Pig и теперь попытаемся загрузить в Pig некоторые данные из локальной файловой системы хоста. Скопируйте с помощью Pig файл из локальной файловой системы в файловую систему HDFS (см. листинг 5).

grunt> mkdir test
grunt> cd test
grunt> copyFromLocal /etc/passwd passwd
grunt> ls
hdfs://0.0.0.0/test/passwd<r 1> 1728

Теперь, когда тестовые данные успешно перенесены в файловую систему Hadoop, можно попробовать создать другой сценарий. Обратите внимание на то, что в Pig можно использовать команду cat для просмотра содержимого файла (просто чтобы убедиться в том, что он на месте). В нашем конкретном примере мы определим количество командных оболочек, назначенных пользователям в файле passwd (последний столбец файла passwd).

Для начала необходимо загрузить файл passwd из файловой системы HDFS в массив Pig. Это делается перед использованием оператора LOAD, но в нашем случае нам нужно разбить столбцы файла passwd на отдельные поля, поэтому мы используем функцию PigStorage, позволяющую указать для файла символ-разделитель (в нашем примере это будет двоеточие [ : ]). Также с помощью ключевого слова AS мы указываем отдельные поля (или схему), включая типы каждого из них (см. листинг 6).

grunt> passwd = LOAD '/etc/passwd' USING PigStorage(':') AS (user:chararray, passwd:chararray, uid:int, gid:int, userinfo:chararray, home:chararray, shell:chararray);
grunt> DUMP passwd;
(root,x,0,0,root,/root,/bin/bash)
(bin,x,1,1,bin,/bin,/sbin/nologin)
...
(cloudera,x,500,500,,/home/cloudera,/bin/bash)
grunt>

Далее с помощью оператора GROUP сгруппируем строки этого массива по версиям содержащихся в них командных оболочек (см. листинг 7). Снова выведем результат на экран, чтобы проиллюстрировать результаты работы оператора GROUP. Обратите внимание на то, что теперь строки сгруппированы (в виде внутреннего массива) по версиям командных оболочек, которые указаны в начале каждой строки.

grunt> grp_shell = GROUP passwd BY shell;
grunt> DUMP grp_shell;
(/bin/bash,{(cloudera,x,500,500,,/home/cloudera,/bin/bash),(root,x,0,0,...), ...})
(/bin/sync,{(sync,x,5,0,sync,/sbin,/bin/sync)})
(/sbin/shutdown,{(shutdown,x,6,0,shutdown,/sbin,/sbin/shutdown)})
grunt>

Однако нашей задачей является получение количества уникальных командных оболочек, содержащихся в файле passwd, и поэтому мы используем оператор FOREACH для выполнения итераций по всем строкам каждой группы, чтобы подсчитать (COUNT), сколько раз встречается каждая командная оболочка (см. листинг 8).

grunt> counts = FOREACH grp_shell GENERATE group, COUNT(passwd);
grunt> DUMP counts;
...
(/bin/bash,5)
(/bin/sync,1)
(/bin/false,1)
(/bin/halt,1)
(/bin/nologin,27)
(/bin/shutdown,1)
grunt>

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

Операторы диагностики

Pig поддерживает несколько операторов диагностики, которые можно использовать для отладки сценариев Pig. Как видно из предыдущих примеров, оператор DUMP оказывается чрезвычайно полезным, если вы хотите увидеть не только данные, но и схему самих данных. Также можно использовать оператор DESCRIBE для получения подробного описания схемы массива (поле и тип).

Оператор EXPLAIN чуть более сложен, но очень полезен. Можно использовать этот оператор для указанного массива, чтобы увидеть, как физические операторы сгруппированы в задания map и reduce (т. е. как были получены данные).

Таблица 2 содержит список операторов диагностики Pig Latin и их описания.

Определяемые пользователями функции

Pig является мощным и полезным языком не только для решения задач, описанных в этой статье; его возможности можно расширять с помощью определяемых пользователями функций (User-Defined Functions, UDF). В сценариях Pig можно использовать ваши собственные функции, разработанные для определенных задач, например, для анализа входных данных или форматирования результирующих данных и даже операторов. Пользовательские функции часто пишутся на языке Java и позволяют Pig поддерживать специализированную обработку, а кроме того, являются способом расширения возможностей Pig для выполнения конкретных задач.

Пользователи Pig

Как видно из этой короткой статьи, Pig - это мощный инструмент для выполнения запросов данных в кластере Hadoop. Этот язык настолько мощный, что по оценкам компании Yahoo! сценарии Pig Latin генерируют от 40% до 60% всей рабочей нагрузки в ее кластерах Hadoop. С учетом того, что в Yahoo! имеется 100 тыс. центральных процессоров и примерно на 50% из них запущен Hadoop, получается, что Pig занимает внушительную долю.

Однако Yahoo! - это не единственная компания, использующая достоинства Pig. Вы можете встретить Pig в таких компаниях, как Twitter (обработка log-файлов и поиск информации в твитах), AOL и MapQuest (анализ и пакетная обработка данных), LinkedIn; во всех этих компаниях Pig используется для поиска людей, которых вы, возможно, знаете. По имеющимся данным компания Ebay использует Pig для оптимизации поиска, а компания adyard использует Pig примерно в половине своих рекомендательных систем.

Что дальше

Для описания всех возможностей Pig по обработке больших объемов данных потребуется отдельная книга книге. Pig упрощает (особенно для людей, не являющимися профессиональными программистами) обработку больших объемов данных в кластерах Hadoop. Изначально Pig был разработан в 2006 году компанией Yahoo! и вскоре после этого был передан Apache Software Foundation, что позволило использовать его в мировом масштабе. Эта передача оказалась результатом осознания разработчиками Yahoo! всей мощи, которую Pig предоставлял людям, не являющимся программистами. По мере того, как Hadoop набирает популярность в качестве инфраструктуры, развивается и экосистема Hadoop, которая изменит представление о больших данных и их постоянно расширяющемся использовании.