Завершая серию статей по IoT-платформе компании Tesla на базе Apache Kafka, сегодня рассмотрим проблемы пиковой загрузки системы и особенности обработки высокоприоритетных событий. Читайте далее, как оптимально определить ключ раздела, чтобы снизить затраты на передачу данных, избежать перегрузки в пиковые моменты и отделить пользователей данных от разработчиков и дата-инженеров.
Тонкости обработки потоков высокоприоритетных событий от устройств интернета вещей
При том, что архитектура IoT-платформы Tesla на базе Apache Kafka кажется достаточно устойчивой, о чем мы писали здесь, здесь и здесь, большое количество конечных устройств, которые отправляют данные в систему, приведет к огромным объемам событий и перегрузке подобно DDoS-атаке. Чтобы снизить расходы на сотовую связь, можно определить Wi-Fi как основной способ передачи данных. Но это чревато пиковой загрузкой в вечернее время, когда пользователи возвращаются с работы и их IoT-устройства начинают загружать данные в домашний Wi-Fi. Подготовка серверов к такой нагрузке может в итоге обойтись компании дороже, чем ожидаемая экономия затрат на связь. При этом важно успешно обрабатывать все потоки сообщений, поскольку некоторые из них могут содержать критически важные данные. В этом случае можно отправлять такие события в конвейер ускоренной обработки, записывая их в соответствующий топик Kafka. Можно также более строго настроить оповещения для этой обработки, чтобы исключить даже малейшее запаздывание.
Добавление отдельной конечной точки в API добавляет сложности, но может позволит быстро поместить высокоприоритетные сообщения в отдельный топик и поток обработки данных. Альтернативным подходом является небольшой промежуточный топик, который просто просматривает метаданные или содержание сообщения, чтобы перенаправить их в соответствующий топик с «сырыми» данными. За это отвечает специальное приложение потоковой передачи событий, которым нужно управлять для поддержки масштабирования и отказоустойчивости. Как именно дата-инженеры Tesla решили эту задачу, мы рассмотрим далее.
Бакетирование времени события от IoT-устройства
После того, как данные поступают в IoT-систему, они разделяются по времени прибытия, чтобы равномерно распределить сообщения по разделам Kafka, гарантируя масштабирование по горизонтали и отсутствие проблем перегрузки в пиковые моменты. При этом необходимо структурировать дальнейшее разделение данных для канонических топиков.
На практике в анализе парка устройств интернета вещей критически важно учитывать временные изменения. Например, чтобы ответить на вопросы «Как изменилась стоимость устройства за последние два дня?» или «Каков средний показатель по всему парку за последний месяц?»
Для некоторых потоков данных можно получить относительно равномерное распределение событий, разделив их по метке времени самого события (event_millisecond). Это не момент, когда сообщение было получено. Разделяя данные по метке времени, события от каждого устройства даже для короткого временного диапазона попадают в отдельный раздел. Это подходит для относительно небольших потоков, где допустимо одновременное использование всех разделов для ответа на вопросы о конкретном устройстве.
Однако, в случае сегментированной базы данных, где хранятся данные потока событий, все не так просто, т.к. разделение, скорей всего, выполнено по тем же ключам, по которым запрашивается IoT-устройство и время. Если несколько приложений пишут данные в базу, чтобы повысить пропускную способность системы с большим количеством разделов, разделение по event_millisecond означает, что каждый раздел Kafka имеет данные для каждого раздела СУБД. Это похоже на то, как непосредственное чтение из топиков Kafka сводится к считыванию сообщений из каждого раздела, чтобы составить небольшое историческое окно для отдельного устройства. При множестве масштабных потоков это довольно сложно и поддерживается далеко не каждой даже масштабируемой СУБД.
Самым простым решением к получению непрерывных потоков событий является разделение по уникальным идентификаторам IoT-устройств (UUID). Это гарантирует, что данные для одного устройства всегда попадают в один и тот же раздел и, кроме повторных попыток и перезапусков, в том же порядке, в котором они были получены. Это упрощает последующую обработку их потребителями, которым нужно будет прочитать всего один раздел, чтобы получить историю устройства.
В то же время есть опасность перегрузки канонических топиков Apache Kafka, когда несколько устройств отправят огромные объемы данных. Это может быть частью обычных бизнес-операций, поскольку устройство подключается к сети и отправляет всю историю, когда оно было отключено или последний сигнал перед выходом устройства из строя, когда оно не сможет обеспечить доступность всей необходимой диагностической информации.
Последовательная схема разделения на основе UUID может привести к однобокому распределению данных по разделам, замедлению записи при перегрузке брокеров и негативному влиянию на пропускную способность последующих потребителей. Впрочем, есть альтернативное решение, вязанное с кейсом экономии средств, рассмотренном в самом начале: маловероятно, что устройства будут постоянно подключены и отправлять поток событий. Поэтому будут существовать отдельные окна событий для каждого сообщения, когда каждое устройство отправляет порцию данных, покрывающую некоторый промежуток времени. Сопоставив эти фрагменты с временными окнами, которые удобны для системы хранения, можно определить ключ раздела канонического топика как комбинацию UUID и time_bucket, где раздел создается динамически в зависимости от времени события.
Чтение сообщений из канонических топиков Apache Kafka
Во время чтения канонических топиков данные IoT-устройств продолжают группироваться в течение длительного времени, избегая риска DDoS-подобных потоков. При большом количестве устройств, которыми нужно управлять и анализировать, важным понимать, сколько из них действительно отправляют данные ежедневно и в какое время суток. Например, в первой половине дня работают 90% всего парка, а вечером – 30%. Если этот показатель внезапно упадет, необходимо оперативно узнать об этом. В таком случае поможет поток метаданных, построенный на необработанном потоке, использованном для синтаксического анализа исходного кода и преобразования его в канонический, что мы разбирали здесь. Эти метаданные включают тип устройства, его UUID, версию прошивки, время получения сообщения, топик источника, размер сообщения и, возможно, время начала/окончания сообщения.
Получив эти данные, можно выполнять мощные аналитические запросы, которые помогут понять состояние IoT-устройств, например, по какой части парка получены данные за определенный период времени, какие устройства никогда не отправляли данные или хронически опаздывают и насколько велики сообщения?
Есть несколько способов материализовать этот поток метаданных для ответа на запросы, например, с помощью другой таблицы в базе данных временных рядов. При этом метаданные в основном генерируются отдельно от типового канонического анализа событий сообщений. Эта работа требует меньше вычислительных затрат с точки зрения ЦП и генерируемых данных, позволяя быстро и дешево отвечать на вопросы о состоянии всех устройств интернета вещей. Поток метаданных не полностью отделяется от всего потока сообщений: когда парсер берет необработанные данные и записывает их в канонический поток, производные метрики дополняют выполняемый анализ. Например, количество событий, проанализированных в сообщении, время синтаксического анализа, успешность синтаксического анализа и пр. Поскольку продюсер Kafka не привязан к одному целевому топику, ему можно передать сообщения из разных топиков, добавляя небольшие по размеру метаданные записи о самом синтаксическом анализе.
Этот прием поможет ответить на вопросы о качестве данных, отправляемых с устройств: как часто они формируются правильно и насколько хорошо пользовательские форматы сжимаются в действительности. Это также полезно в определении степени параллелизма, которую нужно использовать для обработки больших сообщений, которая дает в реальном времени понять длительность синтаксического анализа каждого сообщения, чтобы принимать обоснованные решения относительно пропускной способности по сравнению с дублированием данных во время перезапусков.
Таким образом, обработка метаданных не требует больших затрат на их хранение или вычисление: возрастают некоторые накладные расходы на управление дополнительным потоком, но операционная выгода окупает их. Также это отделяет команды владельцев данных от команд, ответственных за операции и масштабирование потоковой обработки. Внешние разработчики пишут код (парсер) и связывают компоненты вместе, а команда потоковой передачи событий управляет масштабированием ПО и всех операций конвейеров. В результате инфраструктура берет выполняет всю тяжелую работу, обеспечивая пользователям легкость получения метрик и информации о состоянии их конвейеров потоковой передачи событий.
Другие интересные примеры использования Apache Kafka для разработки распределенных приложений потоковой аналитики больших данных вы узнаете на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:
Источники
