Работа с топиками Apache Kafka® с помощью PySpark-заданий в Yandex Data Processing

Кластеры Yandex Data Processing поддерживают интеграцию с кластерами Managed Service for Apache Kafka®. Вы можете записывать сообщения в топики Apache Kafka® и читать сообщения из топиков с помощью PySpark-заданий. При чтении поддерживается пакетная обработка (batch processing) и потоковая обработка (stream processing).

Чтобы настроить интеграцию между кластерами Managed Service for Apache Kafka® и Yandex Data Processing:

1. Подготовьте инфраструктуру.
2. Создайте задания PySpark.

Если созданные ресурсы вам больше не нужны, удалите их.

Необходимые платные ресурсыНеобходимые платные ресурсы

В стоимость поддержки описываемого решения входят:

• Плата за кластер Managed Service for Apache Kafka®: использование вычислительных ресурсов, выделенных хостам (в том числе хостам ZooKeeper), и дискового пространства (см. тарифы Apache Kafka®).
• Плата за кластер Yandex Data Processing (см. тарифы Yandex Data Processing).
• Плата за NAT-шлюз (см. тарифы Virtual Private Cloud).
• Плата за бакет Object Storage: хранение данных и выполнение операций с ними (см. тарифы Object Storage).

Подготовьте инфраструктуруПодготовьте инфраструктуру

Создайте задания PySparkСоздайте задания PySpark

1. На локальном компьютере сохраните следующие скрипты:

   kafka-write.py

   Скрипт для записи сообщений в топик Apache Kafka®:

   #!/usr/bin/env python3
   
   from pyspark.sql import SparkSession, Row
   from pyspark.sql.functions import to_json, col, struct
   
   def main():
      spark = SparkSession.builder.appName("dataproc-kafka-write-app").getOrCreate()
   
      df = spark.createDataFrame([
         Row(msg="Test message #1 from dataproc-cluster"),
         Row(msg="Test message #2 from dataproc-cluster")
      ])
      df = df.select(to_json(struct([col(c).alias(c) for c in df.columns])).alias('value'))
      df.write.format("kafka") \
         .option("kafka.bootstrap.servers", "<FQDN_хоста>:9091") \
         .option("topic", "dataproc-kafka-topic") \
         .option("kafka.security.protocol", "SASL_SSL") \
         .option("kafka.sasl.mechanism", "SCRAM-SHA-512") \
         .option("kafka.sasl.jaas.config",
                 "org.apache.kafka.common.security.scram.ScramLoginModule required "
                 "username=user1 "
                 "password=password1 "
                 ";") \
         .save()
   
   if __name__ == "__main__":
      main()
   

   kafka-read-batch.py

   Скрипт для чтения из топика и для пакетной обработки:

   #!/usr/bin/env python3
   
   from pyspark.sql import SparkSession, Row
   from pyspark.sql.functions import to_json, col, struct
   
   def main():
      spark = SparkSession.builder.appName("dataproc-kafka-read-batch-app").getOrCreate()
   
      df = spark.read.format("kafka") \
         .option("kafka.bootstrap.servers", "<FQDN_хоста>:9091") \
         .option("subscribe", "dataproc-kafka-topic") \
         .option("kafka.security.protocol", "SASL_SSL") \
         .option("kafka.sasl.mechanism", "SCRAM-SHA-512") \
         .option("kafka.sasl.jaas.config",
                 "org.apache.kafka.common.security.scram.ScramLoginModule required "
                 "username=user1 "
                 "password=password1 "
                 ";") \
         .option("startingOffsets", "earliest") \
         .load() \
         .selectExpr("CAST(value AS STRING)") \
         .where(col("value").isNotNull())
   
      df.write.format("text").save("s3a://dataproc-bucket/kafka-read-batch-output")
   
   if __name__ == "__main__":
      main()
   

   kafka-read-stream.py

   Скрипт для чтения из топика и для потоковой обработки:

   #!/usr/bin/env python3
   
   from pyspark.sql import SparkSession, Row
   from pyspark.sql.functions import to_json, col, struct
   
   def main():
      spark = SparkSession.builder.appName("dataproc-kafka-read-stream-app").getOrCreate()
   
      query = spark.readStream.format("kafka")\
         .option("kafka.bootstrap.servers", "<FQDN_хоста>:9091") \
         .option("subscribe", "dataproc-kafka-topic") \
         .option("kafka.security.protocol", "SASL_SSL") \
         .option("kafka.sasl.mechanism", "SCRAM-SHA-512") \
         .option("kafka.sasl.jaas.config",
                 "org.apache.kafka.common.security.scram.ScramLoginModule required "
                 "username=user1 "
                 "password=password1 "
                 ";") \
         .option("startingOffsets", "earliest")\
         .load()\
         .selectExpr("CAST(value AS STRING)")\
         .where(col("value").isNotNull())\
         .writeStream\
         .trigger(once=True)\
         .queryName("received_messages")\
         .format("memory")\
         .start()
   
      query.awaitTermination()
   
      df = spark.sql("select value from received_messages")
   
      df.write.format("text").save("s3a://dataproc-bucket/kafka-read-stream-output")
   
   if __name__ == "__main__":
      main()
   

2. Получите FQDN хоста Apache Kafka® и укажите FQDN в каждом скрипте.

3. Загрузите в корень бакета подготовленные скрипты.

4. Создайте задание PySpark для записи сообщения в топик Apache Kafka®. В поле Main python файл укажите путь до скрипта s3a://dataproc-bucket/kafka-write.py.

5. Дождитесь, когда статус задания изменится на Done.

6. Убедитесь, что данные в топик были успешно записаны. Для этого создайте новое задание PySpark для чтения из топика и для пакетной обработки данных. В поле Main python файл укажите путь до скрипта s3a://dataproc-bucket/kafka-read-batch.py.

7. Дождитесь, когда статус нового задания изменится на Done.

8. Скачайте из бакета файл с прочитанными данными:

   part-00000

   {"msg":"Test message #1 from dataproc-cluster"}
   {"msg":"Test message #2 from dataproc-cluster"}
   

   Файл хранится в новой папке kafka-read-batch-output в бакете.

9. Прочитайте сообщения из топика при потоковой обработке. Для этого создайте еще одно задание PySpark. В поле Main python файл укажите путь до скрипта s3a://dataproc-bucket/kafka-read-stream.py.

10. Дождитесь, когда статус нового задания изменится на Done.

11. Скачайте из бакета файлы с прочитанными данными:

    part-00000

    {"msg":"Test message #1 from dataproc-cluster"}
    

    part-00001

    {"msg":"Test message #2 from dataproc-cluster"}
    

    Файлы хранятся в новой папке kafka-read-stream-output в бакете.

Удалите созданные ресурсыУдалите созданные ресурсы

Некоторые ресурсы платные. Удалите ресурсы, которые вы больше не будете использовать, чтобы не платить за них:

1. Удалите объекты из бакета.

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

   Вручную

   Terraform

   1. Кластер Yandex Data Processing.
   2. Кластер Managed Service for Apache Kafka®.
   3. Бакет.
   4. Группу безопасности.
   5. Подсеть.
   6. Таблицу маршрутизации.
   7. NAT-шлюз.
   8. Сеть.
   9. Сервисный аккаунт.

   1. В терминале перейдите в директорию с планом инфраструктуры.

      Важно

      Убедитесь, что в директории нет Terraform-манифестов с ресурсами, которые вы хотите сохранить. Terraform удаляет все ресурсы, которые были созданы с помощью манифестов в текущей директории.

   2. Удалите ресурсы:

      1. Выполните команду:

         terraform destroy
         

      2. Подтвердите удаление ресурсов и дождитесь завершения операции.

      Все ресурсы, которые были описаны в Terraform-манифестах, будут удалены.