Три узла, один MapReduce: пилот Hadoop на журналах доступа

Hadoop давно мелькал в разговорах про «большие данные», но до реального запуска на живом проекте у нас руки не доходили. На этой неделе дошли. Клиент - интернет-магазин с несколькими годами накопленных журналов доступа nginx, около 400 ГБ в gzip. Вопрос простой: сколько уникальных посетителей (по IP) было за каждый месяц прошлого года? На SQL этот вопрос упёрся бы в то, что лог-файлы не в базе, а в базу их никто никогда не клал. Вот и повод.

Что берём за основу

Вышел Apache Hadoop 1.0.4 - первый по-настоящему стабильный релиз ветки 1.x после долгих месяцев release candidate. Параллельно Cloudera выкатила CDH4 с поддержкой YARN (MRv2), что интересно как архитектурный шаг, но для пилота мы взяли именно «ванильный» Hadoop 1.0.4: понятный, хорошо задокументированный, без лишних движущихся частей.

Три виртуальные машины на нашем стенде: один NameNode + JobTracker, два DataNode + TaskTracker. По 4 ядра и 8 ГБ RAM на каждую, Ubuntu 12.04 LTS. Для пилота - более чем достаточно.

Установка: где теряется время

Установка по официальной документации выглядит прямолинейно, но на практике большую часть времени занимает не установка пакетов, а настройка окружения. Вещи, на которые наступили:

SSH без паролей между всеми узлами. Hadoop требует, чтобы NameNode мог по SSH логиниться на все DataNode без ввода пароля. Звучит банально, но пока генерируешь ключи, раскладываешь authorized_keys и проверяешь что StrictHostKeyChecking no стоит в ssh_config - уходит час.

Java 6 vs Java 7. Hadoop 1.0.4 официально поддерживает Java 6. Java 7 тоже работает, но в некоторых сборках с OpenJDK 7 есть известные проблемы с сериализацией в MapReduce. Взяли Oracle JDK 6 - никаких сюрпризов.

/etc/hosts на всех узлах. HDFS-кластер чувствителен к разрешению имён: если NameNode видит DataNode по одному имени, а DataNode сам себя называет иначе - heartbeat не проходит, узел не присоединяется к кластеру. Полчаса отладки через hdfs dfsadmin -report.

Конфигурационные XML. Пять файлов (core-site.xml, hdfs-site.xml, mapred-site.xml, masters, slaves), которые нужно синхронизировать на все узлы вручную. Пока не поставишь rsync или что-то похожее - каждое изменение конфига превращается в упражнение по вниманию.

После трёх часов возни кластер поднялся. hdfs dfsadmin -report показал два живых DataNode с суммарным местом. Веб-интерфейс NameNode на порту 50070 - тоже живой.

Загрузка данных

Сырые архивы лежали на NAS клиента. Скопировали локально, разархивировали выборку за прошлый год (около 80 ГБ текста), залили в HDFS:

hdfs dfs -mkdir /logs/access
hdfs dfs -put /tmp/access_logs/ /logs/access/

Загрузка 80 ГБ в HDFS заняла около 40 минут - узкое место сеть между стендом и источником, не сам HDFS. Данные разлеглись по двум DataNode, replication factor 2 по умолчанию.

Первый MapReduce-джоб

Задача: для каждого месяца подсчитать количество уникальных IP. Это классика - Map выдаёт пару (YYYY-MM, ip), Reduce считает уникальные значения на ключ.

Писали на Java, потому что Python Streaming тоже работает в Hadoop 1.x, но для первого раза хотелось видеть полный цикл без дополнительного слоя абстракции. Mapper читает строку Combined Log Format, парсит регуляркой, выдаёт ключ-значение. Reducer использует HashSet<String> для подсчёта уникальных.

Запуск:

hadoop jar unique-visitors.jar UniqueVisitors \
  /logs/access /output/unique-visitors

Джоб отработал примерно за 18 минут на всём объёме. Для сравнения: мы прикинули, что последовательный grep + awk на одной машине на том же объёме занял бы в несколько раз больше.

Результат - 12 строк, по одной на месяц, с количеством уникальных IP. Клиент получил ответ на вопрос, который раньше был «технически невозможным».

Что понравилось

HDFS как хранилище для аналитики ведёт себя предсказуемо: файлы лежат, репликация работает, чтение параллельное. Для однократной пакетной обработки больших плоских файлов - хорошо подходит.

MapReduce-модель понятна. После прочтения пары статей и исходника WordCount из примеров Hadoop - написать свой джоб несложно. Это не rocket science, это паттерн.

Веб-интерфейс JobTracker на порту 50030 показывает прогресс задачи в реальном времени: сколько map-задач завершено, сколько reduce, время выполнения каждой. Удобно для первых опытов.

Что раздражает

Verbosity Java API. Количество boilerplate-кода для простого джоба впечатляет. Один класс Mapper, один Reducer, один Driver с конфигурацией - и всё это ради подсчёта уникальных строк. Streaming через Python выглядит привлекательнее для прикладных задач.

Отладка локально сложнее, чем кажется. Запустить джоб в local mode (без кластера) можно, но поведение может отличаться от distributed mode - особенно в части serialization и обработки ошибок. Несколько раз джоб падал только в кластере, в local работал.

Конфигурация XML - это боль. Пять файлов, которые надо синхронизировать вручную, плюс переменные окружения, плюс hadoop-env.sh. При изменении настройки памяти для TaskTracker пришлось перезапускать демоны на всех узлах по очереди. Управление конфигурацией через Puppet или Ansible напрашивается само собой.

Куда смотрим дальше

CDH4 с YARN интересен как следующий шаг: там планировщик ресурсов отделён от MapReduce, что теоретически позволяет запускать не только MR-джобы. Но пока это выглядит как архитектурное усложнение без явной выгоды для наших текущих задач - у нас пакетная аналитика, не стриминг.

Hive мы не трогали в этом пилоте, но следующий запрос клиента уже звучит как «а можно это делать на SQL?» - значит, Hive или что-то похожее неминуемо.

Для проектов BI и DWH где данных много, а структура flat - Hadoop как инструмент первичной обработки и агрегации вполне рабочий. Не замена реляционной базе, не серебряная пуля - просто инструмент для задач, где нужно параллельно обработать сотни гигабайт без предварительной схемы.

Elasticsearch со Logstash мы уже применяли для оперативных логов - там поиск и индексирование в реальном времени. Hadoop - другая история: не поиск, а трансформация и агрегация за один проход. Инструменты не конкурируют, они для разных вопросов.