Working with Apache Kafka® topics using PySpark jobs in Yandex Data Processing

Yandex Data Processing clusters support integration with Managed Service for Apache Kafka® clusters. You can write and read messages to and from Apache Kafka® topics using PySpark jobs. Reading supports both batch processing and stream processing.

To configure integration between Managed Service for Apache Kafka® and Yandex Data Processing clusters:

1. Set up your infrastructure.
2. Create PySpark jobs.

If you no longer need the resources you created, delete them.

Getting startedGetting started

Sign up for Yandex Cloud and create a billing account:

1. Navigate to the management console and log in to Yandex Cloud or create a new account.
2. On the Yandex Cloud Billing page, make sure you have a billing account linked and it has the ACTIVE or TRIAL_ACTIVE status. If you do not have a billing account, create one and link a cloud to it.

If you have an active billing account, you can create or select a folder for your infrastructure on the cloud page.

Learn more about clouds and folders here.

Required paid resourcesRequired paid resources

• Managed Service for Apache Kafka® cluster: use of computing resources allocated to hosts, storage and backup size (see Managed Service for Apache Kafka® pricing).
• Yandex Data Processing cluster: use of computing resources with a Yandex Data Processing markup, use of network drives, retrieval and storage of logs, amount of outgoing traffic (see Yandex Data Processing pricing).
• NAT gateway: hourly use of the gateway and its outgoing traffic (see Yandex Virtual Private Cloud pricing).
• Yandex Object Storage bucket: use of storage, data operations (see Object Storage pricing).

Set up your infrastructureSet up your infrastructure

Create PySpark jobsCreate PySpark jobs

1. On your local computer, save the following scripts:

   kafka-write.py

   Script for writing messages to an Apache Kafka® topic:

   #!/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", "<host_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

   Script for reading from a topic and batch processing:

   #!/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", "<host_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://<new_bucket_name>/kafka-read-batch-output")
   
   if __name__ == "__main__":
      main()
   
   kafka-read-stream.py

   Script for reading from a topic and stream processing:

   #!/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", "<host_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://<new_bucket_name>/kafka-read-stream-output")
   
   if __name__ == "__main__":
      main()
   

2. Get the Apache Kafka® host FQDN and specify it in each script.

3. Upload the prepared scripts to the bucket root.

4. Create a PySpark job for writing a message to the Apache Kafka® topic. In the Main python file field, specify the script path: s3a://<new_bucket_name>/kafka-write.py.

5. Wait for the job status to change to Done.

6. Make sure the data is successfully written to the topic. To do this, create a new PySpark job for reading data from the topic and batch processing. In the Main python file field, specify the script path: s3a://<new_bucket_name>/kafka-read-batch.py.

7. Wait for the new job status to change to Done.

8. Download the file with the read data from the bucket:

   part-00000

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

   The file resides in the new folder named kafka-read-batch-output in the bucket.

9. Read messages from the topic during stream processing. To do this, create another PySpark job. In the Main python file field, specify the script path: s3a://<new_bucket_name>/kafka-read-stream.py.

10. Wait for the new job status to change to Done.

11. Download the files with the read data from the bucket:

    part-00000

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

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

    The files reside in the new folder named kafka-read-stream-output in the bucket.

Delete the resources you createdDelete the resources you created

Some resources are not free of charge. Delete the resources you no longer need to avoid paying for them:

1. Delete the objects from the bucket.

2. Delete the rest of the resources depending on how you created them:

   Manually

   Terraform

   1. Yandex Data Processing cluster.
   2. Managed Service for Apache Kafka® cluster.
   3. Bucket.
   4. Security group.
   5. Subnet.
   6. Route table.
   7. NAT gateway.
   8. Network.
   9. Service account.

   1. In the terminal window, go to the directory containing the infrastructure plan.

      Warning

      Make sure the directory has no Terraform manifests with the resources you want to keep. Terraform deletes all resources that were created using the manifests in the current directory.

   2. Delete resources:

      1. Run this command:

         terraform destroy
         

      2. Confirm deleting the resources and wait for the operation to complete.

      All the resources described in the Terraform manifests will be deleted.