Архитектура Twitter. Два года спустя.

В далеком 2008м я уже публиковал статью про архитектуру Twitter, но время летит стремительно и она уже абсолютно устарела. За это время аудитория Twitter росла просто фантастическими темпами и многое поменялось и с технической точки зрения. Интересно что новенького у одного из самых популярных социальных интернет-проектов?

Статистика

  • 3 год, 2 месяца и 1 день потребовалось Twitter, чтобы набрать 1 миллиард твитов
  • На сегодняшний день, чтобы отправить миллиард твитов пользователям нужна всего одна неделя
  • 752% рост аудитории за 2008 год
  • 1358% рост аудитории за 2009 год (без учета API, по данным comScore)
  • 175 миллионов зарегистрированных пользователей на сентябрь 2010 года
  • 460 тысяч регистраций пользователей в день
  • 9й сайт в мире по популярности (по данным Alexa, год назад был на 12 месте)
  • 50 миллионов твитов в день год назад, 140 миллионов твитов в день месяц назад, 177 миллионов твитов в день на 11 марта 2011г.
  • Рекорд по количеству твитов за секунду 6939, установлен через минуту после того, как Новый Год 2011 наступил в Японии
  • 600 миллионов поисков в день
  • Лишь 25% трафика приходится на веб сайт, остальное идет через API
  • Росто числа мобильных пользователей за последний год 182%
  • 6 миллиардов запросов к API в день, около 70 тысяч в секунду
  • 8, 29, 130, 350, 400 - это количество сотрудников Twitter на январь 2008, январь 2009, январь 2010, январь и март 2011, соответственно

Самая свежая статистика про Twitter.

Платформа

Сравните с аналогичным разделом предыдущей статьи о Twitter - увидите много новых лиц, подробнее ниже.

Оборудование

  • Сервера расположены в NTT America
  • Никаких облаков и виртуализации, существующие решения страдают слишком высокими задержками
  • Более тысячи серверов
  • Планируется переезд в собственный датацентр

Что такое твит?

  • Сообщение длиной до 140 символов + метаданные
  • Типичные запросы:
    • по идентификатору
    • по автору
    • по @упоминаниям пользователей

Архитектура

Процесс обработки запроса в Twitter

Unicorn

Сервер приложений для Rails:

  • Развертывание новых версий кода без простоя
  • На 30% меньше расход вычислительных ресурсов и оперативной памяти, по сравнению с другими решениями
  • Перешли с mod_proxy_balancer на mod_proxy_pass

Rails

Используется в основном для генерации страниц, работа за сценой реализована на чистом Ruby или Scala.

Столкнулись со следующими проблемами:

  • Проблемы с кэшированием, особенно по части инвалидации
  • ActiveRecord генерирует не самые удачные SQL-запросы, что замедляло время отклика
  • Высокие задержки в очереди и при репликации

memcached

  • memcached не идеален. Twitter начал сталкиваться с Segmentation Fault в нем очень рано.
  • Большинство стратегий кэширования основываются на длинных TTL (более минуты).
  • Вытеснение данных делает его непригодным для важных конфигурационных данных (например флагов "темного режима", о котором пойдет речь ниже).
  • Разбивается на несколько пулов для улучшения производительности и снижения риска вытеснения.
  • Оптимизированная библиотека для доступа к memcached из Ruby на основе libmemcached + FNV hash, вместо чистого Ruby и md5.
  • Twitter является одним их наиболее активных проектов, участвующих в разработке libmemcached.

MySQL

  • Разбиение больших объемов данных является тяжелой задачей.
  • Задержки в репликации и вытеснение данных из кэша является причиной нарушения целостности данных с точки зрения конечного пользователя.
  • Блокировки создают борьбу за ресурсы для популярных данных.
  • Репликация однопоточна и происходит недостаточно быстро.
  • Данные социальных сетей плохо подходят для реляционных СУБД:
    • NxN отношения, социальный граф и обход деревьев - не самые подходящие задачи для таких баз данных
    • Проблемы с дисковой подсистемой (выбор файловой системы, noatime, алгоритм планирования)
    • ACID практически не требуется
    • Для очередей также практически непригодны
  • Twitter сталкивался с большими проблемами касательно таблиц пользователей и их статусов
  • Читать данные с мастера при Master/Slave репликации = медленная смерть

FlockDB

Масштабируемое хранилище для данных социального графа:

  • Разбиение данных через Gizzard
  • Множество серверов MySQL в качестве низлежащей системы хранения
  • В Twitter содержит 13 миллиардов ребер графа и обеспечивает 20 тысяч операций записи и 100 тысяч операций чтения в секунду
  • Грани хранятся и индексируются в обоих направлениях
  • Поддерживает распределенный подсчет количества строк
  • Open source!

Среднее время на выполнение операций:

  • Подсчет количества строк: 1мс
  • Временные запросы: 2мс
  • Запись: 1мс для журнала, 16мс для надежной записи
  • Обход дерева: 100 граней/мс

Подробнее про эволюцию систем хранения данных в Twitter в презентации Nick Kallen.

Cassandra

Распределенная система хранения данных, ориентированная на работу в реальном времени:

  • Изначально разработана в Facebook
  • Очень высокая производительность на запись
  • Из слабых сторон: высокая задержка при случайном доступе
  • Децентрализованная, способна переносить сбои оборудования
  • Гибкая схема данных
  • Планируется полный переход на нее по следующему алгоритму:
    • Все твиты пишутся и в Cassandra и в MySQL
    • Динамически часть операций чтения переводится на Cassandra
    • Анализируется реакция системы, что сломалось
    • Полностью отключаем чтение из Cassandra, чиним неисправности
    • Начинаем сначала
  • Обновление: стратегия по поводу использования Cassandra изменилась, попытки использовать её в роли основного хранилища для твитов прекратились, но она продолжает использоваться для аналитики и географической информации.

Подробнее почему Twitter пришел к решению использовать Cassandra можно прочитать в отдельной презентации.

Помимо всего прочего Cassandra планируется использовать используется для аналитики в реальном времени.

Scribe

Пользователи Twitter генерируют огромное количество данных, около 15-25 Гб в минуту, более 12 Тб в день, и эта цифра удваивается несколько раз в год.

Изначально для сбора логов использовали syslog-ng, но он очень быстро перестал справляться с нагрузкой.

Решение нашлось очень просто: Facebook столкнулся с аналогичной проблемой и разработал проект Scribe, который был опубликован в opensource.

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

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

Поддерживаются различные системы для записи в данным,  в том числе обычные файлы и HDFS (о ней ниже).

Этот продукт полностью решил проблему Twitter со сбором логов, используется около 30 различных категорий. В процессе использования была создана и опубликована масса доработок. Активно сотрудничают с командой Facebook в развитии проекта.

Hadoop

Как Вы обычно сохраняете 12Тб новых данных, поступающих каждый день?

Если считать, что средняя скорость записи современного жесткого диска составляет 80Мбайт в секунду, запись 12Тб данных заняла бы почти 48 часов.

На одном даже очень большом сервере данную задачу не решить, логичным решением задачи стало использование кластера для хранения и анализа таких объемов данных.

Использование кластерной файловой системы добавляет сложности, но позволяет меньше заботиться о деталях.

Hadoop Distributed File System (HDFS) предоставляет возможность автоматической репликации и помогает справляться со сбоями оборудования.

MapReduce framework позволяет обрабатывать огромные объемы данных, анализируя пары ключ-значение.

Типичные вычислительные задачи, которые решаются с помощью Hadoop в Twitter:

  • Вычисление связей дружбы в социальном графе (grep и awk не справились бы, self join в MySQL на таблицах с миллиардами строк - тоже)
  • Подсчет статистики (количество пользователей и твитов, например подсчет количества твитов занимает 5 минут при 12 миллиардах записей)
  • Подсчет PageRank между пользователями для вычисления репутации.

В твиттер используется бесплатный дистрибутив от Cloudera, версия Hadoop 0.20.1, данные храняться в сжатом по алгоритму LZO виде, библиотеки для работы с данными опубликованы под названием elephant-bird.

Pig

Для того чтобы анализировать данные с помощью MapReduce обычно необходимо разрабатывать код на Java, что далеко не все умеют делать, да и трудоемко это.

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

Немного напоминает SQL, но намного проще. Это позволяет писать в 20 раз меньше кода, чем при анализе данных с помощью обычных MapReduce работ. Большая часть работы по анализу данных в Twitter осуществляется с помощью Pig.

Данные

Полу-структурированные данные:

  • логи Apache, RoR, MySQL, A/B тестирования, процесса регистрации
  • поисковые запросы

Структурированные данные:

  • Твиты
  • Пользователи
  • Блок-листы
  • Номера телефонов
  • Любимые твиты
  • Сохраненные поиски
  • Ретвиты
  • Авторизации
  • Подписки
  • Сторонние клиенты
  • География

Запутанные данные:

  • Социальный граф

Что же они делают с этим всем?

  • Подсчет математического ожидания, минимума, максимума и дисперсии следующих показателей:
    • Количество запросов за сутки
    • Средняя задержка, 95% задержка
    • Распределение кодов HTTP-ответов (по часам)
    • Количество поисков осуществляется каждый день
    • Количество уникальных запросов и пользователей
    • Географическое распределение запросов и пользователей
  • Подсчет вероятности, ковариации, влияния:
    • Как отличается использование через мобильные устройства?
    • Как влияет использование клиентов сторонних разработчиков?
    • Когортный анализ
    • Проблемы с сайтом (киты и роботы, подробнее ниже)
    • Какие функциональные возможности цепляют пользователей?
    • Какие функциональные возможности чаще используются популярными пользователями?
    • Корректировка и предложение поисковых запросов
    • A/B тестирование
  • Предсказания, анализ графов, естественные языки:
    • Анализ пользователей по их твитам, твитов, на которые они подписаны, твитам их фоловеров
    • Какая структура графа ведет к успешным популярным сетям
    • Пользовательская репутация
    • Анализ эмоциональной окраски
    • Какие особенности заставляют людей ретвитнуть твит?
    • Что влияет на глубину дерева ретвитов ?
    • Долгосрочное обнаружение дубликатов
    • Машинное обучение
    • Обнаружения языка

Подробнее про обработку данных в презентации Kevin Weil.

HBase

Twitter начинают строить настоящие сервисы на основе Hadoop, например поиск людей:

  • HBase используется как изменяемая прослойка над HDFS
  • Данные экспортируются из HBase c помощью периодической MapReduce работы:

На основе HBase разрабатываются и другие продукты внутри Twitter.

Основными её достоинствами являются гибкость и легкая интеграция с Hadoop и Pig.

По сравнению с Cassandra:

  • "Их происхождение объясняет их сильные и слабые стороны"
  • HBase построен на основе системы по пакетной обработке данных, высокие задержки, работает далеко не в реальном времени
  • Cassandra построена с нуля для работы с низкими задержками
  • HBase легко использовать при анализе данных как источник или место сохранения результатов, Cassandra для этого подходит меньше, но они работают над этим
  • HBase на данный момент единственную точку отказа в виде мастер-узла
  • В твиттере HBase используется для аналитики, анализа и создания наборов данных, а Cassandra - для онлайн систем

Loony

Централизованная система управления оборудованием.

Реализована с использованием:

  • Python
  • Django
  • MySQL
  • Paraminko (реализация протокола SSH на Python, разработана и опубликована в opensource в Twitter)

Интегрирована с LDAP, анализирует входящую почту от датацентра и автоматически вносит изменения в базу.

Murder

Система развертывания кода и ПО, основанная на протоколе BitTorrent.

Благодаря своей P2P природе позволяет обновить более тысячи серверов за 30-60 секунд.

Kestrel

Распределенная очередь, работающая по протоколу memcache:

  • set - поставить в очередь
  • get - взять из очереди

Особенности:

  • Отсутствие строгого порядка выполнения заданий
  • Отсутствие общего состояния между серверами
  • Разработана на Scala

Daemon'ы

Каждый твит обрабатывается с помощью daemon'ов.

В unicorn обрабатываются только HTTP запросы, вся работа за сценой реализована в виде отдельных daemon'ов.

Раньше использовалось много разных демонов, по одному на каждую задачу (Rails), но перешли к меньшему их количеству, способному решать несколько задач одновременно.

Как они справляются с такими темпами роста?

Рецепт прост, но эффективен, подходит практически для любого интернет-проекта:

  • обнаружить самое слабое место в системе;
  • принять меры по его устранению;
  • перейти к следующему самому слабому месту.

На словах звучит и правда примитивно, но на практике нужно предпринять ряд мер, чтобы такой подход был бы реализуем:

  • Автоматический сбор метрик (причем в агрегированном виде)
  • Построение графиков (RRD, Ganglia)
  • Сбор и анализ логов
  • Все данные должны получаться с минимальной задержкой, как можно более близко к реальному времени
  • Анализ:
    • Из данных необходимо получать информацию
    • Следить за динамикой показателей: стало лучше или хуже?
    • Особенно при развертывании новых версий кода
    • Планирование использования ресурсов намного проще, чем решение экстренных ситуаций, когда они на исходу

Примерами агрегированных метрик в Twitter являются "киты" и "роботы", вернее их количество в единицу времени.

Что такое "робот"?

Twitter Робот

  • Ошибка внутри Rails (HTTP 500)
  • Непойманное исключение
  • Проблема в коде или нулевой результат
Что такое "кит"?

Twitter Кит

  • HTTP ошибка 502 или 503
  • В твиттер используется фиксированный таймаут в 5 секунд (лучше кому-то показать ошибку, чем захлебнуться в запросах)
  • Убитый слишком длинный запрос к базе данных (mkill)

Значительное превышение нормального количества китов или роботов в минуту является поводом для беспокойством.

Реализован этот механизм простым bash-скриптом, который просматривает агрегированные логи за последние 60 секунд, подсчитывает количество китов/роботов и рассылает уведомления, если значение оказалось выше порогового значения. Подробнее про работу команды оперативного реагирования в презентации John Adams.

"Темный режим"

Для экстренных ситуаций в Twitter предусмотрен так называемый "темный режим", который представляет собой набор механизмов для отключения тяжелых по вычислительным ресурсам или вводу-выводу функциональных частей сайта. Что-то вроде стоп-крана для сайта.

Имеется около 60 выключателей, в том числе и полный режим "только для чтения".

Все изменения в настройках этого режима фиксируются в логах и сообщаются руководству, чтобы никто не баловался.

Подводим итоги

  • Не бросайте систему на самотек, начинайте собирать метрики и их визуализировать как можно раньше
  • Заранее планируйте рост требуемых ресурсов и свои действия в случае экстренных ситуаций
  • Кэшируйте по максимуму все, что возможно
  • Все инженерные решения не вечны, ни одно из решений не идеально, но многие будут нормально работать в течение какого-то периода времени
  • Заранее начинайте задумываться о плане масштабирования
  • Не полагайтесь полностью на memcached и базу данных - они могут Вас подвести в самый неподходящий момент
  • Все данные для запросов в реальном времени должны находиться в памяти, диски в основном для записи
  • Убивайте медленные запросы (mkill) прежде, чем они убьют всю систему
  • Некоторые задачи могут решаться путем предварительного подсчета и анализа, но далеко не все
  • Приближайте вычисления к данным по возможности
  • Используйте не mongrel, а unicorn для RoR

Спасибо за внимание, жду Вас снова! Буду рад, если Вы подпишитесь на меня в Twitter, с удовольствием пообщаюсь со всеми читателями :)