Hadoop 1.0.0 на старом железе: MapReduce на логах ритейл-сайта

Клиент из ритейла позвонил в начале ноября с вопросом, который звучал примерно так: «Мы слышали про Big Data, вы умеете?». Откуда услышал - не уточнил, но скорее всего из какого-то отраслевого мероприятия, где это словосочетание звучало из каждого второго доклада. Конкретная боль была понятна без всяких Big Data: логи веб-сайта накопились за два года, лежат в файлах по 5-20 Гб каждый, и когда маркетинг хочет понять конверсию по источникам трафика за год - аналитик запускает скрипт, который молотит несколько дней, а потом всё равно падает по памяти.

Hadoop 1.0.0 как раз вышел - первый релиз с нормальным номером версии вместо дат и снапшотов. Решили попробовать.

Железо и кластер

Специально покупать ничего не стали. В серверной у клиента нашлись три машины на Intel Xeon, которые раньше были тестовыми серверами: по 16 Гб RAM, SATA-диски на 600-800 Гб, гигабитная сеть. Всё это простаивало. Поставили CentOS 6, Java 6, Hadoop 1.0.0.

Топология минимальная - один NameNode + JobTracker, два DataNode + TaskTracker. NameNode хранит метаданные HDFS и координирует задачи; DataNode - рабочие лошади, где живут данные и выполняются Map/Reduce-таски. Примерно так:

[NameNode / JobTracker]
        |
  [DataNode 1]  [DataNode 2]

Конфигурация hdfs-site.xml и mapred-site.xml - минималистичная. Репликация данных на два (default три - но нод-то только две рабочих). Размер блока HDFS оставили 64 Мб - стандарт для Hadoop 1.x.

Самое неочевидное при первом знакомстве: Hadoop категорически не любит, когда NameNode и DataNode запущены от разных пользователей, когда SSH-ключи между нодами не настроены, и когда часы на серверах расходятся. На последнем потеряли час - DataNode отваливался от кластера с невнятной ошибкой, пока не синхронизировали NTP.

Данные: два года логов nginx

Логи лежали в разных местах: часть на сервере nginx, часть в архивах на NAS, часть вообще на локальных дисках у системного администратора клиента. Собрали в одно место, распаковали, загрузили в HDFS:

hadoop fs -mkdir /logs/nginx
hadoop fs -put /tmp/nginx-logs/*.log /logs/nginx/

Формат лога - стандартный combined: IP, дата, метод, URL, статус, размер, referer, user-agent. Строки вида:

1.2.3.4 - - [01/Jan/2011:00:00:01 +0300] "GET /catalog/shoes HTTP/1.1" 200 12345 "http://yandex.ru/..." "Mozilla/5.0..."

Задача: посчитать уникальные IP по дням, количество запросов по разделам каталога, топ источников трафика (по referer).

Первый MapReduce: подсчёт запросов по разделам

MapReduce - это не просто «параллельный SQL». Нужно думать в терминах Map (каждая строка -> ключ-значение) и Reduce (агрегируем по ключу). Для подсчёта запросов по разделам каталога:

• Map читает строку лога, парсит URL, выдаёт пару (раздел_каталога, 1).
• Reduce суммирует единицы по каждому разделу.

Писали на Java - другого варианта в Hadoop 1.0 нет (Hive и Pig существуют, но мы их не трогали в этот раз). Код немаленький: классы Mapper, Reducer, драйвер, сборка в JAR, запуск через hadoop jar. Для людей, привыкших к SQL, это культурный шок - чтобы сделать GROUP BY COUNT(*), нужно написать страниц пять кода.

Запустили первую задачу. Hadoop поднял прогресс-бар в веб-интерфейсе (порт 50070 для HDFS, 50030 для JobTracker). Через 40 минут - результат в HDFS в виде нескольких файлов part-r-00000, part-r-00001. Забрали, посмотрели - данные сходятся с тем, что аналитик считал вручную по одному из месяцев.

Аналитик клиента, который несколько часов смотрел на прогресс-бар, сказал: «Это долго». Объяснили разницу: его скрипт на Python молотил данные последовательно на одной машине и падал. Hadoop раскидал задачу по нодам, обработал все два года данных и выдал результат. Если добавить ещё нод - будет быстрее.

Что неожиданно сработало и что нет

Сработало хорошо - HDFS как хранилище. Загрузить 200+ Гб логов, убедиться, что данные реплицированы на обе DataNode, и спокойно работать с ними - это работает надёжно. Веб-интерфейс NameNode показывает состояние кластера, репликацию, живые ноды.

Не сработало - рассчитывали, что Java-код будет несложно писать. Это не так. Для каждой новой агрегации нужен отдельный Mapper, отдельный Reducer, новый JAR. Итерации медленные. Если бы задача была разовая - проще было бы написать хороший Python-скрипт с партиционированием.

Неожиданный нюанс - маленькие файлы убивают производительность. У клиента среди логов оказались файлы по 1-5 Мб (ротация каждые полчаса на тихих сайтах). Для HDFS каждый файл - отдельный блок в памяти NameNode. Пришлось предварительно склеивать мелкие файлы командой hadoop fs -cat с перенаправлением в один большой файл.

Что дальше

Кластер из трёх нод сейчас в тестовом режиме. До production далеко: нет нормального мониторинга, нет процедуры загрузки новых логов, нет HA для NameNode (в Hadoop 1.x NameNode - single point of failure). Клиент хочет считать агрегаты раз в неделю - для этого текущей установки достаточно.

На вопрос «когда переходим на Big Data в production?» честный ответ пока такой: посмотрим на практике, сколько реально задач требуют такого подхода. SQL Server справляется с большинством аналитики, когда данные хорошо структурированы. Hadoop - это инструмент для случаев, где данные неструктурированные, объём такой, что не влезает в реляционную БД, или задача натурально параллельная.

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