[Yandex Cloud documentation](../../index.md) > [Tutorials](../index.md) > [Building a data platform](index.md) > Working with Apache Kafka® topics using Yandex Data Processing

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

# 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](../../data-proc/operations/jobs-pyspark.md). 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](#infra).
1. [Create PySpark jobs](#create-jobs).

If you no longer need the resources you created, [delete them](#clear-out).


## Required paid resources {#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](../../managed-kafka/pricing.md)).
* Yandex Data Processing cluster fee (see [Yandex Data Processing pricing](../../data-proc/pricing.md)).
* Fee for a NAT gateway (see [Virtual Private Cloud pricing](../../vpc/pricing.md)).
* Fee for an Object Storage bucket: data storage and operations with it (see [Object Storage pricing](../../storage/pricing.md)).


## Set up your infrastructure {#infra}

{% list tabs group=instructions %}

* Manually {#manual}

   1. [Create a cloud network](../../vpc/operations/network-create.md) named `dataproc-network`, without subnets.
   1. [Create a subnet](../../vpc/operations/subnet-create.md) named `dataproc-subnet-b` in the `ru-central1-b` availability zone.
   1. [Set up a NAT gateway](../../vpc/operations/create-nat-gateway.md) for `dataproc-subnet-b`.
   1. [Create a security group](../../vpc/operations/security-group-create.md) named `dataproc-security-group` in `dataproc-network`.
   1. [Configure the security group](../../data-proc/operations/cluster-create.md#change-security-groups).
   1. [Create a service account](../../iam/operations/sa/create.md) named `dataproc-sa` with the following roles:

      * `storage.viewer`
      * `storage.uploader`
      * `dataproc.agent`
      * `dataproc.user`

   1. [Create a bucket](../../storage/operations/buckets/create.md) named `dataproc-bucket`.
   1. [Grant the `dataproc-sa` service account](../../storage/operations/buckets/edit-acl.md) the `FULL_CONTROL` permission for `dataproc-bucket`.
   1. [Create a Yandex Data Processing cluster](../../data-proc/operations/cluster-create.md#create) with the following parameters:

      * **Cluster name**: `dataproc-cluster`.
      * **Environment**: `PRODUCTION`.
      * **Version**: `2.1`.
      * **Services**:

         * `HDFS`
         * `LIVY`
         * `SPARK`
         * `TEZ`
         * `YARN`

      * **Service account**: `dataproc-sa`.
      * **Availability zone**: `ru-central1-b`.
      * **Bucket name**: `dataproc-bucket`.
      * **Network**: `dataproc-network`.
      * **Security groups**: `dataproc-security-group`.
      * **Subclusters**: Master, one subcluster named `Data` and one subcluster named `Compute`.

   1. [Create a Managed Service for Apache Kafka® cluster](../../managed-kafka/operations/cluster-create.md#create-cluster) with the following parameters:

      * **Cluster name**: `dataproc-kafka`.
      * **Environment**: `PRODUCTION`.
      * **Version**: `3.5`.
      * **Availability zone**: `ru-central1-b`.
      * **Network**: `dataproc-network`.
      * **Security groups**: `dataproc-security-group`.
      * **Subnet**: `dataproc-subnet-b`.

   1. [Create an Apache Kafka® topic](../../managed-kafka/operations/cluster-topics.md#create-topic) with the following parameters:

      * **Name**: `dataproc-kafka-topic`.
      * **Number of partitions**: `1`.
      * **Replication factor**: `1`.

   1. [Create an Apache Kafka® user](../../managed-kafka/operations/cluster-accounts.md#create-user) with the following parameters:

      * **Name**: `user1`.
      * **Password**: `password1`.
      * **Topics the user gets permissions for**: `*` (all topics).
      * **Permissions for the topics**: `ACCESS_ROLE_CONSUMER`, `ACCESS_ROLE_PRODUCER`, and `ACCESS_ROLE_ADMIN`.

* Terraform {#tf}

   1. If you do not have Terraform yet, [install it](../infrastructure-management/terraform-quickstart.md#install-terraform).
   1. [Get the authentication credentials](../infrastructure-management/terraform-quickstart.md#get-credentials). You can add them to environment variables or specify them later in the provider configuration file.
   1. [Configure and initialize a provider](../infrastructure-management/terraform-quickstart.md#configure-provider). There is no need to create a provider configuration file manually, you can [download it](https://github.com/yandex-cloud-examples/yc-terraform-provider-settings/blob/main/provider.tf).
   1. Place the configuration file in a separate working directory and [specify the parameter values](../infrastructure-management/terraform-quickstart.md#configure-provider). If you did not add the authentication credentials to environment variables, specify them in the configuration file.
   1. Download the [kafka-and-data-proc.tf](https://github.com/yandex-cloud-examples/yc-data-proc-spark-kafka/blob/main/kafka-and-data-proc.tf) configuration file to the same working directory.

      This file describes:

      * [Network](../../vpc/concepts/network.md#network).
      * [NAT gateway](../../vpc/concepts/gateways.md) and route table required for Yandex Data Processing.
      * [Subnet](../../vpc/concepts/network.md#subnet).
      * [Security group](../../vpc/concepts/security-groups.md) for the Yandex Data Processing and Managed Service for Apache Kafka® clusters.
      * [Service account](../../iam/concepts/users/service-accounts.md) for the Yandex Data Processing cluster.
      * Service account for managing the Yandex Object Storage bucket.
      * [Yandex Object Storage bucket](../../storage/concepts/bucket.md).
      * [Static access key](../../iam/concepts/authorization/access-key.md) required to grant the service account permissions for the bucket.
      * Yandex Data Processing cluster.
      * Managed Service for Apache Kafka® cluster.
      * Apache Kafka® user.
      * Apache Kafka® topic.

   1. Specify the following in the `kafka-and-data-proc.tf` file:

      * `folder_id`: Cloud folder ID, same as in the provider settings.
      * `dp_ssh_key`: Absolute path to the public key for the Yandex Data Processing cluster. Learn more about connecting to a Yandex Data Processing host over SSH [here](../../data-proc/operations/connect-ssh.md).

   1. Validate your Terraform configuration files using this command:

      ```bash
      terraform validate
      ```

      Terraform will display any configuration errors detected in your files.

   1. Create the required infrastructure:

      1. Run this command to view the planned changes:
      
         ```bash
         terraform plan
         ```
      
         If you described the configuration correctly, the terminal will display a list of the resources to update and their parameters. This is a verification step that does not apply changes to your resources.
      
      1. If everything looks correct, apply the changes:
         1. Run this command:
      
            ```bash
            terraform apply
            ```
      
         1. Confirm updating the resources.
         1. Wait for the operation to complete.

      All the required resources will be created in the specified folder. You can check resource availability and their settings in the [management console](https://console.yandex.cloud).

{% endlist %}

## Create PySpark jobs {#create-jobs}

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

   {% cut "`kafka-write.py`" %}

   Script for writing messages to an Apache Kafka® topic:

   ```python
   #!/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()
   ```

   {% endcut %}

   {% cut "`kafka-read-batch.py`" %}

   Script for reading from a topic and batch processing:

   ```python
   #!/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()
   ```

   {% endcut %}

   {% cut "`kafka-read-stream.py`" %}

   Script for reading from a topic and stream processing:

   ```python
   #!/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()
   ```

   {% endcut %}

1. [Get the Apache Kafka® host FQDN](../../managed-kafka/operations/connect/index.md#get-fqdn) and specify it in each script.
1. [Upload](../../storage/operations/objects/upload.md) the prepared scripts to the bucket root.
1. [Create a PySpark job](../../data-proc/operations/jobs-pyspark.md#create) 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.
1. Wait for the [job status](../../data-proc/operations/jobs-pyspark.md#get-info) to change to `Done`.
1. 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.
1. Wait for the new job status to change to `Done`.
1. [Download](../../storage/operations/objects/download.md) the file with the read data from the bucket:

   {% cut "part-00000" %}

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

   {% endcut %}

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

1. 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.
1. Wait for the new job status to change to `Done`.
1. Download the files with the read data from the bucket:

   {% cut "part-00000" %}

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

   {% endcut %}

   {% cut "part-00001" %}

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

   {% endcut %}

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

{% note info %}

You can view the job logs and search data in them using [Yandex Cloud Logging](../../logging/index.md). For more information, see [Working with logs](../../data-proc/operations/logging.md).

{% endnote %}

## Delete the resources you created {#clear-out}

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

1. [Delete the objects](../../storage/operations/objects/delete.md) from the bucket.
1. Delete the other resources depending on how they were created:

    {% list tabs group=instructions %}

    - Manually {#manual}

        1. [Yandex Data Processing cluster](../../data-proc/operations/cluster-delete.md).
        1. [Managed Service for Apache Kafka® cluster](../../managed-kafka/operations/cluster-delete.md).
        1. [Bucket](../../storage/operations/buckets/delete.md).
        1. [Security group](../../vpc/operations/security-group-delete.md).
        1. [Subnet](../../vpc/operations/subnet-delete.md).
        1. [Route table](../../vpc/operations/delete-route-table.md).
        1. [NAT gateway](../../vpc/operations/delete-nat-gateway.md).
        1. [Network](../../vpc/operations/network-delete.md).
        1. [Service account](../../iam/operations/sa/delete.md).

    - Terraform {#tf}

        1. In the terminal window, go to the directory containing the infrastructure plan.
        
            {% note 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.
        
            {% endnote %}
        
        1. Delete resources:
        
            1. Run this command:
        
                ```bash
                terraform destroy
                ```
        
            1. Confirm deleting the resources and wait for the operation to complete.
        
            All the resources described in the Terraform manifests will be deleted.

    {% endlist %}