[Yandex Cloud documentation](../../index.md) > [Tutorials](../index.md) > [Building a data platform](index.md) > Working with jobs Yandex Data Processing > Working with PySpark jobs

# Working with PySpark jobs

[Apache Spark](https://spark.apache.org/) is a distributed processing framework for unstructured and semi-structured data and a part of the Hadoop project ecosystem.

In this section, we provide a simple example that demonstrates how to use [PySpark](https://spark.apache.org/docs/latest/api/python/), the Spark interface for Python, in Yandex Data Processing. In the example, we use PySpark to count the number of times each word appears in a short text.

To execute a PySpark job:

1. [Set up your infrastructure](#infra).
1. [Create a PySpark job](#create-job).

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


## Getting started {#before-you-begin}

Sign up for Yandex Cloud and create a [billing account](../../billing/concepts/billing-account.md):
1. Navigate to the [management console](https://console.yandex.cloud) and log in to Yandex Cloud or create a new account.
1. On the **[Yandex Cloud Billing](https://center.yandex.cloud/billing/accounts)** page, make sure you have a billing account linked and it has the `ACTIVE` or `TRIAL_ACTIVE` [status](../../billing/concepts/billing-account-statuses.md). If you do not have a billing account, [create one](../../billing/quickstart/index.md) and [link](../../billing/operations/pin-cloud.md) a cloud to it.

If you have an active billing account, you can create or select a [folder](../../resource-manager/concepts/resources-hierarchy.md#folder) for your infrastructure on the [cloud page](https://console.yandex.cloud/cloud).

[Learn more about clouds and folders here](../../resource-manager/concepts/resources-hierarchy.md).


### Required paid resources {#paid-resources}

* 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](../../data-proc/pricing.md)).
* Public IP addresses if public access is enabled for cluster hosts (see [Yandex Virtual Private Cloud pricing](../../vpc/pricing.md)).
* Yandex Object Storage buckets: use of storage, data operations (see [Object Storage pricing](../../storage/pricing.md)).


## Set up your infrastructure {#infra}

1. [Create a service account](../../iam/operations/sa/create.md) with the `dataproc.agent` and `dataproc.provisioner` roles.

1. In Object Storage, [create buckets](../../storage/operations/buckets/create.md) and [configure access](../../storage/operations/buckets/edit-acl.md) to them:
   
   1. Create a bucket for the input data and grant the `READ` permission for this bucket to the cluster service account.
   1. Create a bucket for the processing output and grant the cluster service account `READ and WRITE` permissions for this bucket.

1. [Create a Yandex Data Processing cluster](../../data-proc/operations/cluster-create.md) with the following settings:

    * **Environment**: `PRODUCTION`.
    * **Services**:
        * `HDFS`
        * `SPARK`
        * `YARN`
    * **Service account**: Select the service account you created earlier.
    * **Bucket name**: Select a bucket for the processing results.

## Create a PySpark job {#create-job}

1. Upload a file for processing:
   
   1. Copy and save the following to a file named `text.txt`:
   
       {% cut "text.txt" %}
   
       ```text
       she sells sea shells on the sea shore
       the shells that she sells are sea shells I am sure
       so if she sells sea shells on the sea shore
       I am sure that the shells are sea shore shells
       ```
   
       {% endcut %}
   
   1. [Upload](../../storage/operations/objects/upload.md) the `text.txt` file to the source data bucket.

1. Upload a file with the analysis program code in Python:

    1. Copy and save to the `word_count.py` file:


        {% cut "word_count.py" %}

        ```python
        import sys
        from pyspark import SparkConf, SparkContext


        def main():

            if len(sys.argv) != 3:
                print('Usage job.py <input_directory> <output_directory>')
                sys.exit(1)

            in_dir = sys.argv[1]
            out_dir = sys.argv[2]

            conf = SparkConf().setAppName("Word count - PySpark")
            sc = SparkContext(conf=conf)

            text_file = sc.textFile(in_dir)
            counts = text_file.flatMap(lambda line: line.split(" ")) \
                .map(lambda word: (word, 1)) \
                .reduceByKey(lambda a, b: a + b)

            if out_dir.startswith('s3a://'):
                counts.saveAsTextFile(out_dir) 
            else:
                default_fs = sc._jsc.hadoopConfiguration().get('fs.defaultFS')
                counts.saveAsTextFile(default_fs + out_dir)


        if __name__ == "__main__":
            main()
        ```

        {% endcut %}

    1. [Upload](../../storage/operations/objects/upload.md) the `word_count.py` file to the input data bucket.

1. [Create a PySpark job](../../data-proc/operations/jobs-pyspark.md#create) with the following parameters:

    * **Main python file**: `s3a://<input_data_bucket_name>/word_count.py`
    * **Arguments**:

        * `s3a://<input_data_bucket_name>/text.txt`
        * `s3a://<output_bucket_name>/<output_directory>`

1. Wait for the [job status](../../data-proc/operations/jobs-pyspark.md#get-info) to change to `Done`.

1. [Download](../../storage/operations/objects/download.md) the files with the results from the bucket and review them:

    {% cut "part-00000" %}

    ```text
    ('sea', 6)
    ('are', 2)
    ('am', 2)
    ('sure', 2)
    ```

    {% endcut %}

    {% cut "part-00001" %}

    ```text
    ('she', 3)
    ('sells', 3)
    ('shells', 6)
    ('on', 2)
    ('the', 4)
    ('shore', 3)
    ('that', 2)
    ('I', 2)
    ('so', 1)
    ('if', 1)
    ```

    {% endcut %}

{% 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. To avoid paying for them, delete the resources you no longer need:

1. [Delete the cluster](../../data-proc/operations/cluster-delete.md).
1. [Delete the buckets](../../storage/operations/buckets/delete.md).
1. [Delete the service account](../../iam/operations/sa/delete.md).