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. Prepare your infrastructure.
2. Create PySpark jobs.

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

Prepare the infrastructurePrepare the 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 kafka-read-batch-output folder 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 kafka-read-stream-output folder 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: