В 2012 году социальная сеть Facebook преодолела серьезный рубеж, охватив 1 миллиард пользователей, при этом количество просмотров страниц достигло не менее круглого числа — 1 триллион в месяц. На данный момент хранилище компании составляет около 40 петабайт, ежедневно пользователями генерируется около 90 терабайт новой информации. Сказать, глядя на эти цифры, что Facebook — просто высоконагруженный проект, — значит, ничего не сказать.
Архитектура Facebook представляет собой уникальное решение, которое совмещает целую плеяду известных технологий. Предлагаю рассмотреть общее программное устройство крупнейшей социальной сети мира, а также использовать это как повод для практического знакомства с ее ключевыми технологиями.
Надеюсь, практические примеры инсталляций, приводимые синхронно с изложением сухой теории и статистики, помогут сократить вечный диалектический разрыв между теорией и практикой и в очередной раз подтвердить поговорку «не боги горшки обжигают».
Удивительная масштабируемость применяемых в Facebook решений заключается прежде всего в том, что они одинаково успешно могут быть использованы как в крупных социальных сетях, так и в обыденных проектах средней величины — везде, где требуется повышенная гибкость и скорость реакции системы.
Прежде чем начать, хотелось бы поверхностно перечислить (представить) главные действующие лица, имена которых будут упоминаться по всей статье далее.
Hadoop — это открытая платформа для распределенных вычислений, хранения и обработки «больших данных». Это кросс-платформенный проект фонда Apache Software Foundation, написанный на языке Java, активное участие в разработке которого принимала, в том числе, и Facebook. В последнее время Hadoop стал широко использоваться в высоконагруженных интернет-проектах, для которых требуется масштабируемая и эффективная платформа для массово-параллельной обработки гигантских объемов данных.
Hadoop Project состоит из четырех самостоятельных компонент. Во-первых, это распределенная файловая система HDFS, которая отвечает за хранение «больших данных» на кластере Hadoop. Во-вторых — программная модель (фреймворк), построенная на принципах MapReduce, предназначенная для вычислений и обработки данных на кластере. В качестве третьей составляющей выделяют набор инфраструктурных программных библиотек и утилит для обслуживания Hadoop и смежных проектов. HBase — это заключительная, четвертая часть комплекса, которая представляет из себя нереляционную распределенную базу данных. Это решение — отказоустойчивый способ хранения больших объемов разреженных данных (созданное, кстати, по полной аналогии с BigTable от Google).
Как первая, так и вторая составляющая реализованы на базе ставшей уже стандартной архитектуры ведущий/ведомый (master/slave). В частности, в случае с HDFS ведущий (управляющий метаданными всей файловой системы) сервер называется NameNode, а множество ведомых серверов, физически хранящих данные, — DataNode.
Вторая составляющая (MapReduce) аналогично двулика: она состоит из ведущего планировщика задач JobTracker, который распределяет их по множеству подчиненных ему узлов кластера, каждый из которых называется TaskTracker. Все упомянутые функциональные единицы кластера Hadoop реализованы как независимые демоны.
Замыкает Hadoop-архитектуру Facebook программа Hive. Это надстройка для облегчения программирования типичных аналитических задач на кластере (например, она позволяет использовать непрограммируемые запросы, такие как SQL). Для связывания множества разнородных интерфейсов и унификации разнообразных сервисов и языков программирования используется Thrift — универсальный язык описания интерфейсов. Это высокопроизводительный фреймворк для работы с RPC.
Тесно связана с этим инструментом еще одна часть архитектуры — Scribe, сервер для масштабированного и распределенного ведения логов; также как и Thrift, он был создан компанией Facebook и предназначен для агрегации огромного объема разнородных сообщений, достигающих в случае Facebook нескольких десятков миллиардов в день. Он может как распределять сообщения между разными хранилищами (случайно или в зависимости от хеша), так и дублировать сообщения сразу в несколько хранилищ.
Ниже изображены все этапы обработки входящих запросов социальной сети, вплоть до конечного попадания данных в соответствующие хранилища. Как видно по схеме ниже, критическую роль в этой архитектуре играет технология Hadoop. На текущий момент кластер Facebook Hadoop (Hadoop storage cluster) является самым большим в мире хранилищем данных, построенным на базе этой распределенной технологии.
Инфраструктура Facebook для обработки входящих данных на базе Hadoop (кликабельно)
Вот лишь некоторые актуальные для Facebook цифры, чтобы было понятно, о какой махине далее пойдет речь:
Львиная часть этих данных поступает в кластер через Scribe, который у Facebook реализован не как полностью самостоятельный сервер, а будучи интегрированным в HDFS. Технически это сделано посредством библиотеки libhdfs, которая, по сути, является C-интерфейсом для собственного HDFS-клиента. Такая жесткая монолитная интеграция приводит к переключению работы файловой системы HDFS в режим, близкий к реальному времени.
На самом деле подобный симбиоз не настолько тривиален, как может показаться из этого краткого описания: изначально libhdfs, а также интерфейс FileSystem API содержали огромное количество откровенных ошибок, особенно в многопоточном режиме работы, которые всплыли еще на этапе внедрения в Facebook при тестировании под нагрузкой. Для Scribe многопоточный режим и вовсе используется по умолчанию, из-за чего инженерам Facebook пришлось самостоятельно переработать библиотеку libhdfs, попутно добавив новые вызовы в интерфейсы HDFS (например, важный API-вызов FileSystem.newInstance(), который привнес новый режим подключений к этой файловой системе).
В итоге все данные, оказавшиеся в Hadoop, становятся доступными для обработки через Hive. О важной роли последнего говорит хотя бы тот факт, что примерно 90% всех аналитических задач к кластеру у Facebook генерируется именно фронт-эндом системы Hive.
Хочется упомянуть и стиль сервисного обслуживания подобных монстроидальных информационных систем. В Facebook относительно технического персонала исповедуется принцип «сильной специализации»: над всем программным комплексом работает множество небольших и узкоспециализированных команд, находящихся в двухуровневом подчинении (непосредственный координатор нескольких смежных групп, а также общий для всех комитет развития).
Упомянутый кластер на базе Hadoop/Hive, на хребте которого зиждется работа большинства служб Facebook, в настоящий момент обслуживает всего три человека.
Как и было обещано в начале статьи, будем стараться в качестве познавательного упражнения самостоятельно развертывать описываемые нами системы, и начнем наши практические опыты с компиляции Scribe. Этот инструмент позволяет агрегировать в одном месте-приемнике огромное количество сообщений из самых разных источников: из текстовых файлов, сетевого потока, веб-протоколов, специализированных форматов.
Благодаря стараниям Facebook, отныне доступен режим прямой интеграции с HDFS. Далее мы рассмотрим именно такой вариант, и, в качестве предварительной подсказки, для успешной сборки всех составляющих воедино очень важно правильно подобрать совместимые версии основных пакетов и их зависимостей: boost, libevent, Thrift, fb303 и Hadoop. Наиболее чувствительными компонентами здесь являются boost, который должен иметь версию v2, и Thrift, где подходят любые версии ниже v0.5.0. Конечно, можно выбрать и более свежие версии упомянутого ПО, но при этом нужно быть готовым к сложностям в сборке, решение которых потребует создание собственных патчей (более подробно о специфике этого пути читайте здесь).
Я опускаю тривиальную установку перечисленных выше зависимостей, переходим сразу к сердцу системы — Hadoop/HDFS. Для начала скачиваем и распаковываем последнюю версию Apache Hadoop:
$ wget http://www.us.apache.org/dist/hadoop/common/hadoop-1.1.0/hadoop-1.1.0.tar.gz $ tar xfz hadoop-1.1.0.tar.gz $ cd hadoop-1.1.0/
Перед запуском соблюдаем две важные формальности — указываемым в переменных окружения путь к Java и каталогу для установки Hadoop:
export JAVA_HOME=/System/Library/Frameworks/JavaVM.framework/Versions/1.6.0/Home export HADOOP_INSTALL=/home/savgor/hadoop-1.1.0 export PATH=$PATH:$HADOOP_INSTALL/bin
Запускаем и проверяем правильность работы Hadoop. Получив ожидаемый отклик, переходим к самому интересному — нестандартному «прикручиванию» Scribe. Для начала указываем необходимые пути в переменных окружения LDFLAGS и CPPFLAGS, где должны быть заранее перечислены дополнительные библиотеки. В моем случае эта команда выглядит так:
# export LDFLAGS="-L/usr/local/cdh/hadoop/c++/lib \ # -L/usr/java/default/jre/lib/amd64/server" # export CPPFLAGS="-I/usr/java/default/include \ # -I/usr/local/cdh/hadoop/src/c++/libhdfs \ # -I/usr/java/default/include/linux"
Стандартным образом скачиваем и устанавливаем Scribe, после чего собираем его с ключом —enable-hdfs, последовательно перечисляя локальные пути ко всем зависимостям:
cd scribe-2.2 ./bootstrap.sh --enable-hdfs \ --with-boost=/my_server/boost-1.41.0 \ --with-thriftpath=/my_server/thrift-0.4.0 \ --with-fb303path=/my_server/fb303 \ --prefix=/my_server/scribe-2.2 \ --with-hadooppath=/usr/local/cdh/hadoop make make install
Теперь можно запустить и сам сервер:
bin/scribed -c scribe.conf
Советую сразу же протестировать правильность интеграции Scribe в файловую систему HDFS, для чего можно использовать следующую команду (предварительно нужно установить Scribe-клиент для Python):
python client.py "Igor test string" 100
Альтернативный способ сделать это же — воспользоваться стандартной утилитой пакета Scribe (впрочем, также написанной на Python) scribe_cat так, как показано ниже:
echo “Igor test string” | ./scribe_cat default
Подключиться хранилищу HDFS можно и из других языков, например, Ruby. Более общая информация о тестировании связки Scribe/HDFS доступна здесь, а универсальное решение для связывания хранилища с любыми языками программирования в виде Thrift будет рассмотрено далее во втором примере.
Каким бы способом взаимодействия вы ни воспользовались в итоге, обращаю ваше внимание: Scribe сбрасывает из оперативной памяти данные в HDFS только при накоплении их в объеме, эквивалентном блоку этой файловой системы. Размер нашего тестового сообщения из примера заведомо намного ниже этого значения. Чтобы увидеть результирующий файл, принудительно сбросим данные из буферов сервера, работающего на стандартном порту, следующей командой:
./scribe_ctrl reload 1463
Рассмотреть все богатство возможностей Hadoop/HDFS в столь кратком обзоре невозможно, поэтому в заключение этого пункта приведу выборочно лишь пару повседневных команд, чтобы дать хотя бы поверхностное представление о простоте взаимодействия с Hadoop/HDFS из традиционной консоли.
Для удаления содержимого папки testfiles в распределенном хранилище можно выполнить команду:
/bin/hadoop fs -rm /testfiles/*.*
Пример копирования файла test.txt из локальной файловой системы в HDFS:
/bin/hadoop fs -put /home/user/savgor/test.txt /data
Также полезно знать, что Hadoop/HDFS предоставляет удобную штатную веб-консоль, которая позволяет получать сведения о текущем состоянии системы, и по умолчанию она доступна сразу же после запуска системы по адресу: localhost:50070
.
Интересно, что модель, схематично описанная выше, продержалась в Facebook примерно два года. Из-за постоянного роста нагрузки социальный гигант в данный момент осуществляет миграцию на связку Hbase/HDFS/ZooKepeer тем не менее по-прежнему оставаясь в рамках Hadoop Project. Все сообщения, генерируемые пользователями (приватные сообщения, лайки, чаты, входящие e-mail, SMS и т.д.), — все это агрегируется и накапливается в гигантском хранилище HBase.
На рисунке ниже можно увидеть иерархическую структуру нового трехуровневого кластера HBase/HDFS, к которому в результате вынужденной эволюции пришел Facebook.
Трехуровневая инфраструктура Facebook для обработки данных на базе HBase (кликабельно)
По состоянию на лето 2012 года кластер HBase/HDFS демонстрировал следующие показатели:
Система испытывает чрезвычайную интенсивность ввода/вывода, на этом фоне уже неудивительно смотрится факт, что в дата-центрах Facebook ежемесячно выходят из строя более 1000 винчестеров (есть даже собственный цех для их утилизации).
На фоне этой экстремальной нагрузки очень важно понять причины, по которым именно HBase выбран следующей целью миграции для всей инфраструктуры социального гиганта с Hadoop.
Вот лишь некоторые из них:
Ко всему перечисленному остается добавить, что компания OpenLogic провела исследование тенденций востребованности открытых проектов в 2012 году, и второй год подряд первое место по росту внедрений занимает именно HBase, сразу за ним следует второй наш герой — Hadoop.
Можно констатировать, что фактически Hadoop уже стал именем нарицательным в нереляционном мире Big Data, аналогично SQL в мире обычных РСУБД.
Предлагаю для осмысления всех упомянутых моментов лично попробовать в деле HBase, но чтобы эта задача не показалось слишком банальной — зададимся целью создать к этой базе коннектор для PHP. И сделаем мы это «а-ля Facebook» — через язык Thrift, который позволяет связывать воедино самые различные серверы и языки.
В качестве отправной точки регистрируем все зависимости (libboost-dev, libboost-test-dev, libboost-program-options-dev, libevent-dev), после чего скачиваем и устанавливаем самую последнюю версию Thrift:
$ wget http://dist.apache.org/repos/dist/release/thrift/0.9.0/thrift-0.9.0.tar.gz $ tar xfz thrift-0.9.0.tar.gz $ cd thrift-0.9.0/ $ ./configure $ make $ sudo make install
После инсталляции стоит обратить внимание на итоговый листинг, он будет иметь примерно следующий вид (дается в сокращенном виде):
thrift 0.9.0 Building C++ Library ......... : yes Building C (GLib) Library .... : no Building Java Library ........ : yes Building C# Library .......... : no Building Python Library ...... : yes Building PHP Library ......... : yes Building Erlang Library ...... : no Building TZlibTransport ...... : yes Building TNonblockingServer .. : yes Using ant .................... : /usr/bin/ant Using Python ................. : /usr/bin/python
Если интересующий вас язык отключен, придется выполнить пересборку с явно заданными ключами. Очень частая ошибка при сборке поддержки PHP имеет вид:
configure: error: ./configure failed for lib/php/src/ext/thrift_protocol
В этом случае установите пакет php-dev и повторите попытку сборки. Как видно из листинга выше, в нашем случае PHP благополучно включен, поэтому проверив напоследок работу Thrift (например, запустив команду thrift –version
), переходим к следующему шагу.
Скачиваем и распаковываем последний HBase:
$ wget http://www.eu.apache.org/dist/hbase/hbase-0.94.2/hbase-0.94.2.tar.gz $ tar xfz hbase-0.94.2.tar.gz
Проводим минимальную настройку его конфигурационного XML-файла (conf/hbase-site.xml), чтобы указать путь, где будет храниться наша тестовая база:
<?xml version="1.0"?> <?xml-stylesheet type="text/xsl" href='configuration.xsl'?> <configuration> <property> <name>HBase.rootdir</name> <value>file:///hbase-0.94.2</value> </property> </configuration>
Вот и все приготовления — теперь можно смело запускать HBase:
$ ./hbase-0.94.2/bin/start-hbase.sh
Последний шаг — нужно сгенерировать PHP-код для подключения нашего тестового PHP-клиента. И конечно, сделаем мы это через ранее установленный Thrift:
$ thrift --gen php hbase-0.94.2/src/main/apache/hadoop/hbase/thrift/hbase.thrift
На выходе мы получим папку gen-php, в которой найдем все необходимое для подключения клиента.
В дополнение к нему следует скопировать родные библиотеки Thrift — теперь вы можете использовать этот готовый код для связывания своего проекта с HBase. При использовании своего приложения на базе этого решения не нужно забывать о том, что предварительно у вас должен быть запущен как HBase, так и сервер Thrift (это можно сделать в любой момент, громко скомандовав: /bin/hbase thrift start).
Чтобы сделать пример логически завершенным, проведем краткий тест, который позволит не только проверить работоспособность вышеописанной связки, но и продемонстрирует взаимодействие с установленной нами базой данных. Для этого запустим собственную оболочку HBase:
$ ./hbase-0.94.2/bin/hbase shell Version: 0.92.2, Thu Nov 8 19:25:13 UTC 2012 HBase(main):001:0> create 'Igor', 'hbase', 'test' 0 row(s) in 0.5200 seconds HBase(main):002:0> put 'Igor', 'test!', 'hbase:type1', 'string 1' HBase(main):003:0> put 'Igor', 'test!', 'test:type2', ' string 2' HBase(main):004:0> exit
В этом примере мы создаем таблицу Igor с колонками hbase и test, после чего заполняем их произвольными значениями.
Несмотря на общий обзорный характер статьи, ниже дам немного сжатой информации для тех, кто хочет попробовать Hadoop на практике и всерьёз.
Дефолтную сборку с сайта ASF использовать трудно, гораздо проще брать версии от крупных поставщиков, крупнейшими из которых являются Cloudera и Hortonworks. Самый простой способ — поставить СDH 4.x (Cloudera Distribution of Hadoop) с помощью Cloudera Manager на кластер машин с CentOS 6.2 (это важно — Cloudera Manager работает только с несколькими версиями операционных систем). Кроме полуавтоматической установки Hadoop’а, менеджер позволяет экспортировать клиентскую конфигурацию и обеспечивает мониторинг всего кластера.
Я уже неоднократно говорил: Cloudera для Hadoop это тоже самое, что RedHat — для ядра и Linux; здесь мы имеем гарантированную стабильную работу определенной версии и поддержку патчей для стабильности/безопасности. По количеству коммитов Клоудера больше пилит mr и немного hdfs, основной вклад Хортонворка идет в hdfs ну в попытку пропихнуть YARN в массы. Саму Клоудеру я очень уважаю, так как почти все продукты, что они разрабатывают, в итоге отдают в Apache, хотя и продолжают развивать их.
Как уже частично было описано выше, в Hadoop всего 2 основные части — HDFS и MapReduce-фреймворк. Всё остальное — логические надстройки над этими компонентами.
Отмечу, что наравне с Hadoop 1.x сейчас развивается Hadoop 2.x, где MapReduce переведён на новую платформу YARN, обеспечивающую более рациональное использование ресурсов кластера. В YARN больше нет упомянутых выше JobTracker и TaskTrackers, а вместо этого используется Application Master и Containers. Однако YARN только-только вышел в бету после нескольких лет альфы, поэтому его стабильность пока вызывает вопросы.
Помимо YARN в Hadoop 2.x добавили пачку плюшек и для HDFS, из них особенно доставляют:
Кстати, задачи MapReduce можно писать не только на Java, но и на C++ через Hadoop Pipes или любом другом языке (Python, Ruby, Bash, etc.) — через Streaming API.
Scribe — чисто фейсбуковская игрушка, но на практике гораздо удобней использовать Apache Flume (c HDFS Sink). Hive — относительно медленный и поддерживает только небольшую часть SQL. У Cloudera есть нечто похожее — Impala, которая поддерживает ещё меньшую часть SQL, но зато работает на порядок быстрей.
Вообще говоря — Hive, повторюсь, достаточно медленный, потому как использует MapReduce в качестве своего основного способа работы. А тут и оказывается, что Map все результаты кидает на диск после окончания, а Reduce подымает все эти данные с диска и только потом с ними как-то натужно работает, и если запрос пришёл достаточно большой, то может оказаться, что там несколько MapReduce-операций разом, что еще больше замедляет запрос. Для борьбы с этим можно посмотреть на интересный проект Tez, который пилит Hortonworks, и хотя он потребует YARN на кластере, но в результате у вас будет что-то вроде направленного потока вычислений — вот чуть подробнее об этом.
Impala же по своей сути больше похожа на teradata или greenplum (распределенная выборка, сортировка, p2p-коммуникация между нодами, распределенный hash join и т.д.). Для batch-процессинга лучше подходит Hive, для realtime-выборок из малых и средних объемов — заруливает Impala.
Scalability и reliability в Hadoop отличные. Для NameNode можно настроить standby-сервер, при этом данные будут литься сразу в оба сервера (не путать standby-сервер с Secondary NameNode, который просто немного разгружает NameNode, выполняя за него некоторые вторичные операции). Если оперативной памяти NameNode не хватает для хранения всей метаинформации, обычно используют NameNode Federation и отдают каждому «неймноду» только несколько разделов от всей файловой системы. Задачи MapReduce также могут повторяться несколько раз для обеспечения выполнения даже при перебоях выполнения.
Но, отдельно отмечу — проблема масштабирования дополнительно имеет корни еще и в самой архитектуре Namenode:
~
В заключении рекомендую эту отличную презентацию суммирующую всё вышесказанное о новых фичах Hadoop 2.x:
Если у вас есть время, можно также посмотреть весьма информативное видео на сегодняшнюю тему. Это немного староватая (за 2011 год) презентация именно про «Hadoop Scalability at Facebook» — там все данные непосредственно от первоисточника, поэтому весьма интересно. У них NameNode уже тогда была на 100Gb, отчего начинала упираться в стратосферу, да и холодный старт «так себе».
Вообщем, чего я пересказываю — это видео с «YaC2011. Доклады», там и как они HA достигают, и как NameNode скалят. Проблему масштабирования они (судя по презентации) решили, но в лоб. Отмасштабировали SPoF — вместо одной большой и 100%, в итоге получили 50 маленьких и по 2%. Это выступление идёт на русском языке продолжительностью 40 минут и называется «Масштабируемость Hadoop в Facebook — Дмитрий Мольков, Facebook»:
Facebook и его дата-центр хорош скорее для Северной Америки, но у нас размещение серверов или колокейшн лучше делать в дата-центре United DC (это так называемый colocation или колокейшн). Имейте в виду, что качественный хостинг и размещение серверов для любых сервисов очень важно для будущего проекта.
PDF-файл слайдов данной презентации/выступления удобно смотреть параллельно с видео — вот линк для его скачки + whitepaper с более подробным разъяснением того, что рассказывал выше о Hadoop Дмитрий.