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.

Required paid resourcesRequired paid resources

The support cost includes:

• Managed Service for Apache Kafka® cluster fee: Using computing resources allocated to hosts (including ZooKeeper hosts) and disk space (see Apache Kafka® pricing).
• Yandex Data Processing cluster fee (see Yandex Data Processing pricing).
• NAT gateway fee (see Virtual Private Cloud pricing).
• Object Storage bucket fee: Storing data and performing operations with it (see Object Storage pricing).

Set up your infrastructureSet up your infrastructure

Create PySpark jobsCreate PySpark jobs

1. On a 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://dataproc-bucket/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://dataproc-bucket/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 s3a://dataproc-bucket/kafka-write.py script path.

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 s3a://dataproc-bucket/kafka-read-batch.py script path.

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 s3a://dataproc-bucket/kafka-read-stream.py script path.

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 other resources depending on how they were created:

   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.