Я начал миссию по интеграции Apache Flink с Kafka и PostgreSQL, используя Docker. Что делает это предприятие особенно захватывающим, так это использование pyFlink — Python-версии Flink, которая одновременно мощная и относительно редкая. Эта настройка направлена на эффективную обработку и хранение данных в реальном времени. В следующих разделах я продемонстрирую, как я этого добился, обсуждая возникшие проблемы и способы их преодоления. В конце я предоставлю пошаговое руководство, чтобы вы могли создать и экспериментировать с этим потоком данных самостоятельно.
Настройка инфраструктуры
Инфраструктура, которую мы создадим, иллюстрируется ниже. Внешне есть модуль издателя, который симулирует сообщения датчиков IoT, аналогично тому, что обсуждалось в предыдущем посте. Внутри Docker-контейнера мы создадим две темы Kafka. Первая тема, sensors, будет хранить входящие сообщения от IoT-устройств в реальном времени. Приложение Flink затем будет потреблять сообщения из этой темы, фильтровать те, у которых температура выше 30°C, и публиковать их во вторую тему, alerts. Кроме того, приложение Flink будет вставлять потребленные сообщения в таблицу PostgreSQL, созданную специально для этой цели. Эта настройка позволяет сохранять данные датчиков в структурированном табличном формате, предоставляя возможности для дальнейшей трансформации и анализа. Инструменты визуализации, такие как Tableau или Power BI, могут быть подключены к этим данным для построения графиков и панелей в реальном времени.
Более того, тему alerts могут потреблять другие клиенты для инициирования действий на основе содержащихся в ней сообщений, таких как активация систем кондиционирования воздуха или запуск протоколов пожарной безопасности.
Проблемы с портами Kafka в docker-compose.yml
Изначально я столкнулся с проблемами конфигурации портов Kafka при использовании образа Docker от
Причиной сбоя было то, что внутренние и внешние хосты использовали один и тот же порт, что приводило к проблемам с подключением. Я исправил это, изменив внутренний порт на 19092. Я нашел этот пост в блоге довольно разъясняющим:
- KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:19092,PLAINTEXT_HOST://localhost:9092
Конфигурация Flink в режиме сессии
Для запуска Flink в режиме сессии (позволяющем выполнять несколько задач в одном кластере) я использую следующие директивы в
Пользовательский образ Docker для PyFlink
Учитывая ограничения стандартного образа Docker для PyFlink...