Интеграция Apache Flink с Kafka и PostgreSQL с использованием Docker

Я начал миссию по интеграции 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 от confluentinc, популярного выбора для таких настроек. Эта проблема стала очевидной через логи, подчеркивая важность не запускать docker-compose up в режиме отсоединения (-d) на начальных этапах настройки и устранения неполадок.

Причиной сбоя было то, что внутренние и внешние хосты использовали один и тот же порт, что приводило к проблемам с подключением. Я исправил это, изменив внутренний порт на 19092. Я нашел этот пост в блоге довольно разъясняющим: