фон

Поскольку платформа Kafka привлекает все больше и больше клиентов, я обнаружил, что многие люди знают только Kafka. Тема может запускать механизм очистки данных на основе настройки размера сохранения и времени сохранения, но я не знаком с Kafka. Тема Еще одна компактная стратегия уборки. Поэтому эта статья, которая в основном знакомит с принципом компактности, связана с Конфигурация, Практические. иногда записи операций, соответствующий анализ исходного кода и т. д. Добро пожаловать в фокус общедоступный аккаунт WeChat: Специалисты по работе с большими данными

Компактный принцип

Стратегия очистки данных Kafka определяется параметром log.cleanup.policy. В настоящее время поддерживаются две стратегии: удаление (по умолчанию для обычных тем) и сжатие (по умолчанию для системных тем). Обе стратегии могут использоваться одновременно, не конфликтуя друг с другом. Таким образом, для log.cleanup.policy можно настроить удаление или сжатие или удаление, сжатие. В этой статье на данный момент не рассматривается стратегия очистки при удалении, а только стратегия компактной очистки. Стратегия очистки по умолчанию системной темы Kafka __consumer_offsets компактна.

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

Мы знаем, что тема состоит из раздела. Производитель записывает сообщения в раздел, и каждому сообщению присваивается уникальное и неизменяемое смещение. Как показано ниже:

Если политика очистки — удаление, то механизм очистки данных запускается при выполнении условий размера сохранения или времени сохранения. Сообщения до указанного смещения будут удалены, то есть сообщения до точки хранения удаления, как показано на следующем рисунке:

Другими словами, стратегия удаления не учитывает ключ или значение сообщения, не говоря уже о том, существует ли сообщение с тем же ключом. Компактная стратегия будет рассматривать сообщения с одинаковым ключом в одном и том же разделе и в конечном итоге сохранять только сообщение, соответствующее последнему значению, среди сообщений с тем же ключом. Как показано на рисунке ниже, K1 содержит три сообщения в исходных данных. После компактной обработки сохраняется только одно сообщение K1:V4. Лично я считаю, что называть этот процесс компактным не очень уместно. Его следует называть обновлением или чем-то в этом роде.

Стратегия «Компакт» подходит для сценариев, в которых вы хотите сохранить только текущий снимок, а не полную историю изменений. Например: чтобы сохранить информацию о зарплате сотрудника, вы можете создать тему «зарплата сотрудника» и установить компактную политику, как показано на следующем рисунке:

Гарантия CompactKey

1. Не влияет на потребителей, у которых нет задержки потребления для получения всех сообщений. Другими словами, Compact будет работать только на неактивных сегменте, а потребитель без задержки потребления активно потребляет segment。    

2. Compact не изменит значение смещения, значение ключа, значение раздела и последовательность сообщений. Просто удалите несколько сообщений.

3. В log.cleaner.delete.retention.ms (по умолчанию 24 часа) потребители по-прежнему могут использовать сообщения для удаления. 

В дополнение к обычным сообщениям, содержащим ключ и значение, в Compact также есть специальное сообщение: ключ нормальный, но значение = null. Это сообщение называется сообщением-захоронением. Бессмысленно объединять сообщения-захоронения, поэтому при компактировании такие сообщения будут удалены. Добро пожаловать на подписку на публичный аккаунт: специалисты по работе с большими данными

Компактная конфигурация

log.cleanup.policy: политика очистки (удалить или сжать или удалить, сжать).

log.cleaner.enable: включать ли задачи компактной очистки.

log.cleaner.threads: количество задач компактной очистки.

log.segment.bytes: Максимальное количество байтов файла segmemnt.

log.segment.ms: максимальное время, в течение которого сегмент остается активным.

log.cleaner.backoff.ms: Задача очистки, время простоя, время сна

log.cleaner.min.compaction.lag.ms: минимальное время задержки для запуска сжатия.

log.cleaner.max.compaction.lag.ms: максимальное время задержки для запуска компакта

log.cleaner.dedupe.buffer.size: задача очистки памяти, используемой для дедупликации.

log.cleaner.delete.retention.ms: компактное время задержки удаления сообщения об удалении    

log.cleaner.io.buffer.load.factor: задача очистки нить IO скорость загрузки буфера

log.cleaner.io.buffer.size: задача очистки нитIO буферная память

log.cleaner.io.max.bytes.per.sec: ограничение скорости очистки нитIO

log.cleaner.min.cleanable.ratio: Коэффициент загрязнения, при котором происходит сжатие

Практические случаи

1. Создайте тестовую тему

Чтобы облегчить тестирование, установите min.cleanable.dirty.ratio=0,001 и сегмент.ms=5000, чтобы гарантировать, что задача компактной очистки будет выполнена как можно скорее, и установите разделы=1, чтобы гарантировать, что тестовые сообщения будут записываться в один и тот же раздел. .

2. Опишите тему теста

3. Запустите потребителей

4. Запустите продюсер и отправьте тестовые сообщения.

В содержимое сообщения намеренно добавляются повторяющиеся ключи, а именно:

Patrick,salary: 10000

Lucy,salary: 20000

Bob,salary: 20000

Patrick,salary: 25000

Lucy,salary: 30000

Patrick,salary: 30000

5. Просмотрите сообщения о потреблении на шаге 3.

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

6. Подождите одну минуту и ​​продолжайте выдавать сообщения, например: Стефан, зарплата: 0

7. Запустите новых потребителей

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

Анализ исходного кода

KafkaServer.startup запустит LogManager, а LogManager.startup запустит пул потоков расписания и LogCleaner (CleanerThread запускается внутри). В пуле потоков расписания есть задача kafka-log-retention, которая соответствует стратегии очистки удаления, а LogCleaner соответствует стратегии компактной очистки.

В этой статье описывается только компактный LogCleaner. Его метод запуска следующий:

/**

   * Start the background cleaning

   */

  def startup() {

    info("Starting the log cleaner")

// Количество очистителей передается через параметр log.cleaner.threadsКонфигурация, по умолчанию — 1.

    (0 until config.numThreads).foreach { i =>

      val cleaner = new CleanerThread(i)

      cleaners += cleaner

      cleaner.start()

    }

  }

Далее мы в основном рассмотрим основной процесс класса CleanerThread, который находится в файле LogCleaner.scala.

Основной процесс заключается в следующем:

doWork -> cleanFilthiestLog ->  grabFilthiestCompactedLog -> cleanLog -> clean -> doClean

Из-за ограниченного места содержимое метода doClean представлено не будет. Пожалуйста, прочитайте сами, если вам интересно. следующее:

https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/log/LogCleaner.scala#L594

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

Подвести итог

Прочитав эту статью, вы сможете освоить принципы компактности, настройку, практические операции, анализ исходного кода и т. д. На этом этапе Kafka Compact Topic легко использовать и настраивать!