Apache Kafka. Потоковая обработка и анализ данных. Apache Kafka. Потоковая обработка и анализ данныхСерия Бестселлеры OReilly
 Скачать 7.59 Mb.
|
|
128 Глава 5 • Внутреннее устройство Kafka Как.мы.уже.обсуждали,.запрос.должен.быть.отправлен.ведущим.репликам.указан- ных.в.запросе.разделов,.для.чего.клиенты.предварительно.запрашивают.метадан- ные,.чтобы.гарантировать.правильную.маршрутизацию.запросов.на.извлечение.. Ведущая.реплика,.получив.запрос,.первым.делом.проверяет,.корректен.ли.он.—. существует.ли.по.крайней.мере.данное.смещение.в.этом.разделе?.Если.клиент. запрашивает.настолько.старое.смещение,.что.оно.уже.удалено.из.раздела,.или.еще. не.существующее,.брокер.вернет.сообщение.об.ошибке. Если.смещение.существует,.брокер.читает.сообщения.из.раздела.вплоть.до.ука- занного.клиентом.в.запросе.ограничения.и.отправляет.сообщения.клиенту..Kafka. знаменита.своим.использованием.метода. zero-copy .для.отправки.сообщений. клиентам.—.это.значит,.что.она.отправляет.сообщения.напрямую.из.файлов.(или,. скорее,.кэша.файловой.системы.Linux).без.каких-либо.промежуточных.буферов.. В.большинстве.же.баз.данных,.в.отличие.от.Kafka,.перед.отправкой.клиентам. данные.сохраняются.в.локальном.буфере..Эта.методика.позволяет.избавиться.от. накладных.расходов.на.копирование.байтов.и.управление.буферами.памяти.и.су- щественно.повышает.производительность. Помимо.ограничения.сверху.объема.возвращаемых.брокером.данных.клиенты. могут.задать.и.ограничение.снизу..Например,.ограничение.снизу.в.10.Кбайт. эквивалентно.указанию.брокеру.возвращать.результаты.только.при.накоплении. хотя.бы.10.Кбайт.для.отправки..Это.отличный.способ.снижения.загруженности. процессора.и.сети.в.случаях,.когда.клиенты.читают.данные.из.тем.с.не.слишком. большими.объемами.трафика..Вместо.отправки.брокерам.запросов.данных.каж- дые.несколько.секунд.с.получением.в.ответ.лишь.одного-двух.(а.то.и.ни.одного). сообщений.клиент.отправляет.запрос,.а.брокер.ждет,.пока.не.накопится.поря- дочный.объем.данных,.возвращает.их,.и.лишь.тогда.клиент.запрашивает.новые. данные.(рис..5.3)..При.этом.читается.в.целом.тот.же.самый.объем.данных.при. намного.меньшем.объеме.взаимодействий,.а.следовательно,.меньших.накладных. расходах. Рис. 5.3. Брокер откладывает ответ до тех пор, пока не накопит достаточно данных Обработка запросов 129 Интересно.отметить,.что.не.все.данные.из.ведущей.реплики.в.этом.разделе.доступны. клиентам.для.считывания..Большинство.клиентов.могут.читать.только.те.сообщения,. которые.записаны.во.все.согласованные.реплики.(ведомых.реплик.это.не.касается,. хотя.они.и.потребляют.данные,.иначе.не.смогла.бы.функционировать.репликация).. Мы.уже.обсуждали,.что.ведущая.реплика.раздела.знает,.какие.сообщения.были. реплицированы.на.какие.реплики,.и.сообщение.не.будет.отправлено.потребителю,. пока.оно.не.записано.во.все.согласованные.реплики..Попытки.прочитать.подобные. сообщения.приведут.к.возврату.пустого.ответа,.а.не.сообщения.об.ошибке. Причина.в.том,.что.не.реплицированные.на.достаточное.количество.реплик.со- общения.считаются.небезопасными.—.в.случае.аварийного.сбоя.ведущей.реплики. и.ее.замены.другой.они.пропадут.из.Kafka..Если.разрешить.клиентам.чтение.со- общений,.имеющихся.только.на.ведущей.реплике,.возникнет.рассогласованность.. Например,.если.во.время.чтения.потребителем.такого.сообщения.ведущая.реплика. аварийно.прекратит.работу,.а.этого.сообщения.нет.больше.ни.на.одном.брокере,. то.оно.будет.утрачено..Больше.ни.один.потребитель.его.прочитать.не.сможет,.что. приведет.к.рассогласованию.с.уже.прочитавшим.его.потребителем..Вместо.этого. необходимо.дождаться.получения.сообщения.всеми.согласованными.репликами. и.лишь.затем.разрешать.потребителям.его.читать.(рис..5.4)..Такое.поведение. означает.также,.что.в.случае.замедления.по.какой-либо.причине.репликации. между.брокерами.доставка.новых.сообщений.потребителям.будет.занимать.боль- ше.времени,.поскольку.мы.сначала.ждем.репликации.сообщений..Эта.задержка. ограничивается.параметром. replica.lag.time.max.ms .—.промежутком.времени,. по.истечении.которого.реплика,.отстающая.в.репликации.новых.сообщений,.будет. сочтена.рассогласованной. Рис. 5.4. Потребители видят только те сообщения, которые реплицированы на согласованные реплики Другие запросы Мы.только.что.обсудили.самые.распространенные.типы.запросов,.применяемые. клиентами.Kafka:. Metadata ,. Produce .и. Fetch ..Важно.помнить,.что.мы.говорим.про. обобщенный.двоичный.протокол,.используемый.клиентами.для.взаимодействия. по.сети..Хотя.Kafka.включает.Java-клиенты,.реализованные.и.поддерживаемые. 130 Глава 5 • Внутреннее устройство Kafka участниками.проекта.Apache.Kafka,.существуют.также.клиенты.на.иных.языках. программирования,.например,.C,.Python,.Go.и.многих.других..Полный.список. можно.найти.на.сайте.Apache.Kafka.( http://www.bit.ly/2sKvTjx ),.причем.все.они.взаи- модействуют.с.брокерами.Kafka.по.этому.протоколу. Кроме.того,.этот.же.протокол.применяется.для.взаимодействия.между.самими. брокерами.Kafka..Эти.запросы.носят.внутренний.характер.и.не.должны.исполь- зоваться.клиентами..Например,.контроллер,.извещая.о.новой.ведущей.реплике. раздела,.отправляет.запрос. LeaderAndIsr .новой.ведущей.реплике,.чтобы.она.начала. принимать.запросы.клиентов,.и.ведомым.репликам,.чтобы.они.ориентировались. на.новую.ведущую.реплику. Протокол.Kafka.в.настоящий.момент.включает.более.20.типов.запросов,.в.даль- нейшем.будут.добавлены.и.другие..Этот.протокол.постоянно.совершенствует- ся.—.он.должен.развиваться.по.мере.добавления.функциональных.возможно- стей.клиентов..Например,.в.прошлом.потребители.Kafka.задействовали.Apache. ZooKeeper.для.отслеживания.получаемых.от.платформы.смещений..Так.что. потребитель.после.запуска.может.выяснить.в.ZooKeeper,.каково.последнее.про- читанное.из.соответствующих.разделов.смещение,.и.будет.знать,.с.какого.места. начинать.обработку..По.различным.причинам.мы.решили.перестать.использо- вать.для.этого.ZooKeeper,.а.хранить.смещения.в.отдельной.теме.Kafka..Для.этого. пришлось.добавить.в.протокол.несколько.типов.запросов:. OffsetCommitRequest ,. OffsetFetchRequest .и. ListOffsetsRequest ..Теперь,.когда.приложение.обращается. к.API.клиента. commitOffset() ,.клиент.ничего.не.записывает.в.ZooKeeper,.а.отправ- ляет.запрос. OffsetCommitRequest .в.Kafka. Темы.по-прежнему.создаются.с.помощью.утилит.командной.строки,.непосред- ственно.обновляющих.список.тем.в.ZooKeeper,.а.брокеры.по.этому.списку.от- слеживают.добавление.новых.тем..Мы.находимся.в.процессе.усовершенствования. Kafka.и.добавления.типа.запроса. CreateTopicRequest ,.с.помощью.которого.все.кли- енты.(даже.в.языках.программирования,.где.нет.библиотеки.ZooKeeper).могли.бы. создавать.темы,.непосредственно.обращаясь.к.брокерам.Kafka. Помимо.усовершенствования.протокола.путем.добавления.новых.типов.запросов. мы.иногда.меняем.существующие.запросы,.добавляя.некоторые.новые.возмож- ности..Например,.при.переходе.от.Kafka.0.9.0.к.0.10.0.мы.решили,.что.клиентам. не.помешает.информация.о.текущем.контроллере,.и.добавили.ее.в.ответ. Metadata В.результате.появилась.новая.версия.запроса.и.ответа. Metadata ..Теперь.клиен- ты.0.9.0.отправляют.запросы. Metadata .версии.0.(поскольку.версия.1.в.них.еще. не.существовала),.а.брокеры.вне.зависимости.от.их.версии.возвращают.ответ.вер- сии.0,.в.котором.нет.информации.о.контроллере..Это.нормально,.ведь.клиенты.0.9.0. не.ждут.информации.о.контроллере.и.все.равно.не.сумеют.выполнить.ее.синтакси- ческий.разбор..Клиент.же.0.10.0.отправит.запрос. Metadata .версии.1,.в.результате. чего.брокеры.версии.0.10.0.вернут.ответ.версии.1.с.информацией.о.контроллере,. которой.клиенты.0.10.0.смогут.воспользоваться..Брокер.же.версии.0.9.0.при.полу- чении.от.клиента.0.10.0.запроса. Metadata .версии.1.не.будет.знать,.что.с.ним.делать,. и.вернет.сообщение.об.ошибке..Именно.поэтому.мы.рекомендуем.сначала.обновить. Физическое хранилище 131 все.брокеры,.а.только.потом.обновлять.клиенты.—.новые.брокеры.смогут.обрабо- тать.старые.запросы,.но.не.наоборот. В.версии.Kafka.0.10.0.мы.добавили.запрос. ApiVersionRequest ,.позволяющий. клиентам.запрашивать.у.брокера.поддерживаемые.версии.запросов.и.применять. соответствующую.версию..Клиенты,.правильно.реализующие.эту.новую.возмож- ность,.смогут.взаимодействовать.со.старыми.брокерами.благодаря.использованию. поддерживаемых.ими.версий.протокола. Физическое хранилище Основная.единица.хранения.Kafka.—.реплика.раздела..Разделы.нельзя.разносить. по.нескольким.брокерам.или.даже.по.различным.дискам.одного.брокера,.так.что. размер.раздела.ограничивается.доступным.на.отдельной.точке.монтирования.ме- стом..(Точка.монтирования.может.состоять.или.из.отдельного.диска.при.исполь- зовании.дискового.массива.JBOD,.или.из.нескольких.дисков,.когда.задействуется. RAID.—.см..главу.2.) При.настройке.Kafka.администратор.задает.список.каталогов.для.хранения.раз- делов.с.помощью.параметра. log.dirs .(не.путайте.его.с.местом.хранения.журнала. ошибок.Kafka,.настраиваемым.в.файле. log4j.properties )..Обычная.конфигурация. включает.по.каталогу.для.каждой.точки.монтирования.Kafka. Разберемся,.как.Kafka.использует.доступные.каталоги.для.хранения.данных.. Во-первых,.посмотрим,.как.данные.распределяются.по.брокерам.в.кластере.и.ка- талогам.на.брокере..Затем.рассмотрим,.как.брокеры.обращаются.с.файлами,.уделив. особое.внимание.гарантиям.сохранения.информации..Затем.заглянем.внутрь. файлов.и.изучим.форматы.файлов.и.индексов..Наконец,.обсудим.сжатие.журна- лов.—.продвинутую.возможность,.благодаря.которой.Kafka.превращается.в.долго- временное.хранилище.данных,.и.опишем,.как.она.функционирует. Распределение разделов При.создании.темы.Kafka.сначала.принимает.решение.о.распределении.разделов.по. брокерам..Допустим,.у.нас.есть.6.брокеров.и.мы.хотим.создать.тему.на.10.разделов. с.коэффициентом.репликации.3..Kafka.нужно.распределить.30.реплик.разделов.по. 6.брокерам..Основные.задачи.этого.распределения.следующие. Равномерно.распределить.реплики.по.брокерам.—.в.нашем.случае.выделить.на. каждый.брокер.5.разделов. Гарантировать,.что.все.реплики.для.каждого.из.разделов.находятся.на.разных. брокерах..Если.ведущая.реплика.раздела.0.располагается.на.брокере.2,.ее.ве- домые.реплики.можно.поместить.на.брокеры.3.и.4,.но.не.на.2.(и.не.обе.на.3). Если.у.брокеров.имеется.информация.о.размещении.в.стойках.(доступны. в.Kafka.начиная.с.версии.0.10.0),.то.желательно.по.возможности.разместить. 132 Глава 5 • Внутреннее устройство Kafka реплики.для.каждого.из.разделов.на.различных.стойках..Это.гарантирует,.что. отсутствие.связи.или.неработоспособность.целой.стойки.не.приведет.к.полной. недоступности.разделов. Чтобы.добиться.этого,.мы.начнем.с.произвольного.брокера.(допустим,.4).и.станем. циклически.назначать.разделы.каждому.из.брокеров.для.определения.местополо- жения.ведущих.реплик..Так,.ведущая.реплика.раздела.0.окажется.на.брокере.4,. ведущая.реплика.раздела.1.—.на.брокере.5,.раздела.2.—.на.брокере.0.(поскольку. у.нас.всего.6.брокеров).и.т..д..Далее.для.каждого.из.разделов.будем.размещать. реплики.в.соответствии.со.все.увеличивающимися.смещениями.по.отношению. к.ведущей.реплике..Если.ведущая.реплика.раздела.0.находится.на.брокере.4,.то. первая.ведомая.реплика.попадет.на.брокер.5,.а.вторая.—.на.брокер.0..Ведущая.ре- плика.раздела.1.находится.на.брокере.5,.так.что.первая.ведомая.реплика.попадет. на.брокер.0,.а.вторая.—.на.брокер.1. Если.учитывать.информацию.о.стойках,.то.вместо.выбора.брокеров.в.числовом. порядке.подготовим.список.брокеров.с.чередованием.стоек..Допустим,.нам.извест- но,.что.брокеры.0,.1.и.2.находятся.в.одной.стойке,.а.брокеры.3,.4.и.5.—.в.другой.. Вместо.того.чтобы.подбирать.брокеры.по.порядку.от.0.до.5,.мы.упорядочиваем. их.в.следующем.порядке:.0,.3,.1,.4,.2,.5.—.за.каждым.брокером.следует.брокер.из. другой.стойки.(рис..5.5)..В.таком.случае,.если.ведущая.реплика.раздела.0.находит- ся.на.брокере.4,.то.первая.реплика.окажется.на.брокере.2,.находящемся.в.другой. стойке..Это.замечательно,.потому.что,.если.первая.стойка.выйдет.из.строя,.у.нас. все.равно.окажется.работающая.реплика.и.раздел.по-прежнему.будет.доступен.. Это.справедливо.для.всех.реплик,.так.что.мы.гарантировали.доступность.в.случае. отказа.одной.из.стоек. Рис. 5.5. Разделы и реплики распределяются по брокерам, находящимся в разных стойках Физическое хранилище 133 Выбрав.нужные.брокеры.для.всех.разделов.и.реплик,.мы.должны.определиться. с.каталогом.для.новых.разделов..Сделаем.это.отдельно.для.каждого.раздела.. Принцип.очень.прост:.подсчитывается.число.разделов.в.каждом.каталоге.и.новые. разделы.добавляются.в.каталог.с.минимальным.числом.разделов..Это.значит,.что. при.добавлении.нового.диска.все.новые.разделы.будут.создаваться.на.нем,.по- скольку.до.того,.как.все.выровняется,.на.новом.диске.всегда.будет.меньше.всего. разделов. Не забывайте про место на диске Обратите внимание на то, что распределение разделов по брокерам не учитывает наличие места или имеющуюся нагрузку, а распределение разделов по дискам учитывает только число разделов, но не их размер. Так что если на некоторых брокерах больше дискового пространства, чем на других (допустим, из-за того что в кластере есть и более старые, и более новые серверы), а среди разделов попадаются очень большие или на брокере есть диски разного размера, необхо- димо соблюдать осторожность при распределении разделов. Управление файлами Сохранение.информации.имеет.в.Kafka.большое.значение.—.платформа.не.хранит. данные.вечно.и.не.ждет,.когда.все.потребители.прочтут.сообщение,.перед.тем.как. его.удалить..Администратор.Kafka.задает.для.каждой.темы.срок.хранения.—.или. промежуток.времени,.в.течение.которого.сообщения.хранятся.перед.удалением,. или.объем.хранимых.данных,.по.исчерпании.которого.старые.сообщения.удаля- ются. Поскольку.поиск.сообщений,.которые.нужно.удалить,.в.большом.файле.и.по- следующее.удаление.его.части.—.процесс.затратный.и.грозящий.возникновением. ошибок,.то.вместо.этого.разделы.разбиваются.на.сегменты..По.умолчанию.каждый. сегмент.содержит.1.Гбайт.данных.или.данные.за.неделю.в.зависимости.от.того,. что.оказывается.меньше..По.достижении.этого.лимита.при.записи.брокером.Kafka. данных.в.раздел.файл.закрывается.и.начинается.новый. Сегмент,.в.который.в.настоящий.момент.производится.запись,.называется.ак- тивным (active.segment)..Активный.сегмент.никогда.не.удаляется,.так.что.если. в.конфигурации.журналов.задано.хранить.данные.лишь.за.день,.но.каждый.сегмент. содержит.данные.за.пять.дней,.то.в.действительности.будут.храниться.данные.за. пять.дней,.поскольку.удалить.их.до.закрытия.сегмента.невозможно..Если.вы.ре- шите.хранить.данные.неделю.и.создавать.новый.сегмент.каждый.день,.то.увидите,. что.каждый.день.будет.создаваться.новый.сегмент.и.удаляться.наиболее.старый,. так.что.почти.все.время.раздел.будет.насчитывать.семь.сегментов. 134 Глава 5 • Внутреннее устройство Kafka Как.вы.знаете.из.главы.2,.брокеры.Kafka.держат.открытыми.дескрипторы.файлов. для.всех.сегментов.раздела,.даже.неактивных..Из-за.этого.число.открытых.де- скрипторов.файлов.стабильно.высоко,.так.что.операционная.система.должна.быть. настроена.соответствующим.образом. Формат файлов Каждый.сегмент.хранится.в.отдельном.файле.данных,.в.котором.находятся.сооб- щения.Kafka.и.их.смещения..Формат.файла.на.диске.идентичен.формату.сообще- ний,.отправляемых.от.производителя.брокеру,.а.затем.от.брокера.потребителям.. Одинаковый.формат.данных.на.диске.и.передаваемых.дает.Kafka.возможность. использовать.оптимизацию.zero-copy.при.передаче.сообщений.потребителям,. а.также.избежать.распаковки.и.повторного.сжатия.данных,.уже.сжатых.произ- водителем. Каждое.сообщение.помимо.ключа,.значения.и.смещения.содержит.такие.вещи,. как.размер.сообщения,.контрольную.сумму.для.обнаружения.порчи.данных,.вол- шебный.байт.с.версией.формата.сообщения,.кодек.сжатия.(Snappy,.GZip.или.LZ4). и.метку.даты/времени.(добавлена.в.версии.0.10.0)..В.зависимости.от.настроек. метка.даты/времени.присваивается.или.производителем.при.отправке.сообщения,. или.брокером.при.его.получении. Если.производитель.отправляет.сообщения.в.сжатом.виде,.то.все.сообщения. из.одного.пакета.сжимаются.вместе.и.отправляются.в.виде.значения.сообще- ния-адаптера.(рис..5.6)..Так.что.брокер.получает.одно.сообщение,.которое.затем. переправляет.потребителю..Но.потребитель,.распаковав.значение.этого.сообще- ния-адаптера,.увидит.все.содержащиеся.в.пакете.сообщения.с.их.метками.даты/ времени.и.смещениями. Это.значит,.что.при.задействовании.сжатия.на.производителе.(что.рекомендует- ся!).отправка.бо'льших.пакетов.приводит.к.более.высокой.степени.сжатия.как.при. передаче.по.сети,.так.и.при.размещении.на.дисках.брокера..Это.также.значит,.что. при.изменении.используемого.потребителями.формата.сообщений.(например,. добавлении.в.сообщение.метки.даты/времени).придется.поменять.как.формат. данных.на.диске,.так.и.протокол.передачи.данных,.а.брокеры.Kafka.должны.будут. знать,.что.делать.в.случаях,.когда.из-за.появления.новых.версий.файлы.содержат. сообщения.в.двух.форматах. Брокеры.Kafka.идут.в.комплекте.с.утилитой. DumpLogSegment ,.позволяющей.про- сматривать.сегменты.разделов.в.файловой.системе.и.исследовать.их.содержимое.. Она.выводит.смещение,.контрольную.сумму,.волшебный.байт,.размер.и.кодек.сжатия. для.каждого.сообщения..Запустить.ее.можно.с.помощью.следующей.команды: bin/kafka-run-class.sh kafka.tools.DumpLogSegments 136 Глава 5 • Внутреннее устройство Kafka Если.задать.параметр. --deep-iteration ,.утилита.отобразит.информацию.о.сжатых. сообщениях,.содержащихся.внутри.сообщений-адаптеров. Индексы Kafka.дает.потребителям.возможность.извлекать.сообщения,.начиная.с.любого. смещения..Это.значит,.что,.если.потребитель.запрашивает.1.Мбайт.сообщений,. начиная.со.смещения.100,.брокер.сможет.быстро.найти.сообщение.со.смещени- ем.100.(которое.может.оказаться.в.любом.из.сегментов.раздела).и.начать.с.этого. места.чтение.сообщений..Чтобы.ускорить.поиск.брокерами.сообщений.с.заданным. смещением,.Kafka.поддерживает.индексы.для.всех.разделов..Индекс.задает.соот- ветствие.смещения.файлу.сегмента.и.месту.в.этом.файле. Индексы.также.разбиты.на.сегменты,.так.что.при.очистке.старых.сообщений. можно.удалять.и.старые.записи.индексов..Kafka.не.поддерживает.для.индексов. контрольные.суммы..В.случае.повреждения.индекс.восстанавливается.из.соот- ветствующего.сегмента.журнала.с.помощью.обычного.повторного.чтения.сообще- ний.и.записи.смещений.и.местоположений..При.необходимости.администраторы. могут.без.опасений.удалять.сегменты.индексов.—.они.будут.сгенерированы.заново. автоматически. Сжатие При.обычных.обстоятельствах.Kafka.хранит.сообщения.в.течение.заданного. интервала.времени.и.удаляет.сообщения,.чей.возраст.превышает.срок.хранения.. Однако.представьте.себе,.что.вы.применяете.Kafka.для.хранения.адресов.доставки. покупателей..В.этом.случае.имеет.смысл.хранить.последний.адрес.каждого.из.по- купателей,.а.не.адреса.за.последнюю.неделю.или.последний.год..Так.вам.не.нужно. будет.волноваться.об.устаревших.адресах,.и.адреса.тех.покупателей,.которые.давно. никуда.не.переезжали,.станут.храниться.столько,.сколько.нужно..Другой.сценарий. использования.—.приложение,.использующее.Kafka.для.хранения.своего.текущего. состояния..При.каждом.изменении.состояния.приложение.записывает.свое.новое. состояние.в.Kafka..При.восстановлении.после.сбоя.оно.читает.эти.сообщения.из. Kafka.для.восстановления.последнего.состояния..В.таком.случае.приложение. интересует.только.последнее.состояние.перед.сбоем,.а.не.все.происходившие.во. время.его.работы.изменения. Kafka.поддерживает.подобные.сценарии.за.счет.двух.возможных.стратегий.со- хранения.для.темы:.delete.(удалять),.при.которой.события,.чей.возраст.превышает. интервал.хранения,.удаляются,.и.compact.(сжимать),.при.которой.сохраняется. только.последнее.значение.для.каждого.из.ключей.темы..Конечно,.вторая.стратегия. имеет.смысл.только.для.тех.тем,.для.которых.приложения.генерируют.события,. содержащие.как.ключ,.так.и.значение..В.случае.использования.неопределенных. ключей.( null ).попытка.сжатия.приведет.к.сбою. |