Обработка потоковых данных Managed Kafka с помощью Managed Flink

Перед началом работы

1. Зарегистрируйтесь в личном кабинете Cloud.ru.

   Если вы уже зарегистрированы, войдите под своей учетной записью.

2. Создайте бакет Object Storage с названием labs.

3. Создайте кластер Data Platform, в котором будет размещен инстанс.

4. Скачайте и установите root-сертификат на устройство.

5. Создайте кластер Managed Kafka.

6. В Managed Kafka создайте топики input_lab1_agg и output_lab1_agg.

7. Создайте секрет с логином и паролем от Managed Kafka в формате JSON:

   {
      "user": "<admin_login>"
      "password": "<admin_password>"
   }
   
   

8. Создайте инстанс Managed Flink.

   На шаге Общие параметры:

   • В поле Бакет выберите labs.

   • Активируйте опцию Добавить секрет и выберите тот, в котором сохранены логин и пароль от Managed Kafka.

   • Выберите Подсеть, в которой расположен кластер Managed Kafka.

9. Сохраните скрипт агрегации данных из потока.

   В параметре properties.bootstrap.servers измените значение <your_kafka_ip> на внутренний IP кластера Managed Kafka вида 10.0.0.0:1234. Параметр встречается два раза. Измените значение в каждом.

10. Загрузите скрипт flink-kafka-stream в бакет Object Storage labs.

1. Загрузите данные в Managed Kafka

На этом шаге вы загрузите данные в Kafka для имитации потока данных.

1. Подключитесь к Managed Kafka.

2. Добавьте сообщения:

   Входные данные для Managed Kafka

   Должно получиться девять сообщений в топике input_lab1_agg.

2. Создайте задачу в Managed Flink

На этом шаге вы запустите задачу.

1. Создайте задачу, заполнив поля:

   • Тип запускаемой задачи — Python.

   • Путь к запускаемому файлу — выберите скрипт flink-kafka-stream.

2. Остальные поля заполните на свое усмотрение.

3. Нажмите Создать.

Дождитесь, когда задача перейдет в статус «Выполняется».

3. Проверьте результаты

На этом шаге вы проверите результат выполнения задачи Managed Flink.

1. Чтобы посмотреть отправленные сообщения, запустите команду в отдельном терминале:

   kcat -C \
        -b <внутренний_IP_брокера>:<порт_брокера> \
        -t <имя_топика> \
        -X security.protocol=SASL_PLAINTEXT \
        -X sasl.mechanism=SCRAM-SHA-512 \
        -X sasl.username="<логин_пользователя>" \
        -X sasl.password="<пароль_пользователя>" \
        -f 'Key: %k\nValue: %s\n'
   
   

   В ответе вернутся сообщения с полем window_end, в котором находится значение закрытия временного окна:

   {
     "type": "temperature",
     "window_indications": 15.5666666668,
     "indications_type": "celsius"
     "window_end": "2026-01-01 12:00:00"
   }
   
   

2. Чтобы проверить обработку данных в реальном времени, отправьте новые данные в топик input_lab1_agg.

3. Проверьте сообщения.

   В ответе появятся новые сообщения с другим временем в поле window_end.

Результат

Вы запустили задачу Managed Flink по обработке потоков данных из Managed Kafka.