Просмотр сообщений из топика Managed Kafka® в логах задач Managed Spark

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

1. Зарегистрируйтесь в личном кабинете Cloud.ru. Если вы уже зарегистрированы, войдите под своей учетной записью.

2. Создайте бакет Object Storage, в котором будут храниться необходимые файлы и логи.

3. Настройте DNS-сервер и подсеть.

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

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

6. Создайте пароль и добавьте его в Secret Manager. Этот секрет станет паролем для доступа к интерфейсу Managed Spark.

7. Создайте инстанс Managed Spark.

8. Убедитесь, что в проекте, где будет запускаться задача Managed Spark, доступен сервис Managed Kafka®.

9. Создайте кластер Managed Kafka®. На шаге Сетевые настройки в списке Подсеть выберите подсеть, указанную при создании инстанса Managed Spark.

10. Подключитесь к кластеру Managed Kafka® и отправьте несколько сообщений в топик.

1. Подготовьте скрипт задачи

1. Скопируйте скрипт и назовите файл kafka_spark.py.

   from pyspark.sql import SparkSession
   import os
   
   kafka_user = os.environ["KAFKA_USER"]
   kafka_pass = os.environ["KAFKA_PASS"]
   kafka_topic = os.environ["KAFKA_TOPIC"]
   kafka_server = os.environ["KAFKA_SERVER"]
   
   spark = SparkSession.builder.appName("kafka").getOrCreate()
   
   df = (
      spark.read.format("kafka")
      .option("kafka.bootstrap.servers", kafka_server)
      .option("kafka.security.protocol", "SASL_PLAINTEXT")
      .option("kafka.sasl.mechanism", "SCRAM-SHA-512")
      .option(
         "kafka.sasl.jaas.config",
         f'org.apache.kafka.common.security.scram.ScramLoginModule required username="{kafka_user}" password="{kafka_pass}";',
         )
      .option("subscribe", kafka_topic)
      .option("startingOffsets", "earliest")
      .option("endingOffsets", "latest")
      .load()
   )
   
   df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
   df.show(truncate=False)
   spark.stop()
   
   

2. Скопируйте скрипт и назовите файл kafka_spark_streaming.py.

   from pyspark.sql import SparkSession
   import os
   
   kafka_user = os.environ["KAFKA_USER"]
   kafka_pass = os.environ["KAFKA_PASS"]
   kafka_topic = os.environ["KAFKA_TOPIC"]
   kafka_server = os.environ["KAFKA_SERVER"]
   
   spark = (
      SparkSession.builder.appName("kafka")
      .getOrCreate()
   )
   
   df = (
      spark
      .readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", kafka_server)
      .option("kafka.security.protocol", "SASL_PLAINTEXT")
      .option("kafka.sasl.mechanism", "SCRAM-SHA-512")
      .option(
         "kafka.sasl.jaas.config",
         f'org.apache.kafka.common.security.scram.ScramLoginModule required username="{kafka_user}" password="{kafka_pass}";',
         )
      .option("subscribe", kafka_topic)
      .option("startingOffsets", "earliest")
      .load()
      )
   
   df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
   сonsole = (df
      .writeStream
      .outputMode('append')
      .format('console')
      .start()
      )
   console.awaitTermination()
   spark.stop()
   
   

3. Откройте ранее созданный бакет Object Storage.

4. Загрузите файлы со скриптами.

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

1. Перейдите в сервис Managed Spark.

2. Откройте созданный ранее инстанс.

3. Перейдите на вкладку Задачи.

4. Нажмите Создать задачу.

5. В блоке Общие параметры введите название задачи, например kafka-spark-streaming.

6. В блоке Образ выберите базовый образ Spark-3.5.0.

7. В блоке Скрипт приложения:

   • В поле Тип запускаемой задачи выберите Python.

   • В поле Путь к запускаемому файлу укажите путь к файлу kafka_spark.py.

8. В блоке Настройки активируйте опцию Добавить параметры окружения. Добавьте следующие параметры и их значения:

   Параметр	Значение
   KAFKA_USER	Логин для подключения к кластеру Managed Kafka®, например, cloud-admin.
   KAFKA_PASS	Пароль для подключения к кластеру Managed Kafka® с указанным логином.
   KAFKA_TOPIC	Имя топика Managed Kafka®.
   KAFKA_SERVER	Внутренний IP-адрес кластера Managed Kafka®.

   Чтобы узнать внутренний IP-адрес, логин и пароль, откройте сервис Managed Kafka® в отдельной вкладке, в списке кластеров нажмите на название созданного ранее кластера и перейдите в блок Данные для подключения.

9. В блоке Настройки активируйте опцию Добавить Spark конфигурацию (–conf).

   • В поле Аргумент укажите spark.jars.packages.

   • В поле Значение укажите org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0.

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

3. Проверьте логи

1. В строке задачи нажмите и выберите Перейти к логам.

2. Используйте фильтр, чтобы найти логи, содержащие сообщения из топика Managed Kafka®.

4. Запустите непрерывное чтение топика Managed Kafka®

На этом шаге вы создадите вторую задачу Managed Spark с использованием скрипта, который будет непрерывно поддерживать соединение с топиком Managed Kafka® и выполнять чтение поступающих в него сообщений.

1. В строке задачи Managed Spark, выполненной ранее, нажмите и выберите Скопировать задачу.

2. В блоке Скрипт приложения в поле Путь к запускаемому файлу укажите путь к файлу kafka_spark_streaming.py.

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

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

5. В строке задачи нажмите и выберите Перейти к логам.

6. Используйте фильтр, чтобы найти логи, содержащие сообщения из топика Managed Kafka®. Если в топик Managed Kafka® не поступают новые данные, в логах будут только отправленные ранее сообщения и информация об ожидании.

7. Отправьте новое сообщение в топик Managed Kafka®.

8. Посмотрите в логах задачи Managed Spark информацию о новом сообщении.

9. Задача Managed Spark c данными параметрами будет выполняться, пока вы ее не завершите. Чтобы завершить задачу, в строке задачи нажмите и выберите Остановить.

Результат