[Yandex Cloud documentation](../../index.md) > [Yandex DataSphere](../index.md) > [Concepts](index.md) > DataSphere Notebook > Computing on Apache Spark™ clusters > Computing on Apache Spark™ clusters

# Computing on Apache Spark™ clusters

To work with Yandex Data Processing clusters, a project needs to be pre-configured. For more information, see [this concept](data-processing.md#settings).

Regardless of the deployment option, all Yandex Data Processing clusters are charged based on the [Yandex Data Processing pricing policy](../../data-proc/pricing.md). To view all the clusters available in your project, navigate to **Project resources** ⟶ ![image](../../_assets/data-processing/data-processing.svg) **Yandex Data Processing** on the project page.

{% note warning %}

As a user of a cluster deployed in Yandex Data Processing, you manage its lifecycle yourself. Even with no computations ongoing, the cluster will not be deleted and will remain billable.

{% endnote %}

DataSphere supports using Yandex Data Processing clusters via:

 * [Spark connector](#spark-with-existing-cluster)
 * [Livy session](#livy-sessions)

## Spark connectors {#spark-with-existing-cluster}

A [Spark connector](spark-connector.md) is a special resource that stores connection settings for Yandex Data Processing clusters. Cluster connection settings are specified when [creating a Spark connector](../operations/data/spark-connectors.md#create). The clusters you select are either connected or created when you run computations in a cell.

You can [publish](../operations/data/spark-connectors.md#create) a Spark connector in a community, thus making it available for other projects. Changes to a Spark connector's settings will apply to all the projects the connector is used in.

For correct integration with DataSphere via a Spark connector, make sure that the [image version](../../data-proc/concepts/environment.md) of the deployed Yandex Data Processing cluster is at least `2.0`, with `LIVY`, `SPARK`, and `YARN` enabled.

To learn more about using Spark connectors, see [this guide](../operations/data/spark-connectors.md).

### Running Python code in a cluster {#run-code-with-spark-connector}

1. Enter Python code in a cell. Here is an example:

   ```python
   import random

   def inside(p):
       x, y = random.random(), random.random()
       return x*x + y*y < 1

   NUM_SAMPLES = 1_000_000_0

   count = sc.parallelize(range(0, NUM_SAMPLES)).filter(inside).count()
   print("Pi is roughly %f" % (4.0 * count / NUM_SAMPL))
   ```
1. Select the code cell and run your computations.
1. In the **Notebook VM configurations** window that opens, go to the **With Yandex Data&nbsp;Processing cluster** tab.
1. Select a computing resource configuration and a Spark connector.
1. Click **Select**.

You set the connector and environment configuration manually only when running a computation for the first time. All subsequent computations in this notebook will be performed on the VM created during the first run. A Yandex Data Processing cluster connects to the VM using the Spark connector.

### Syncing Python environment with a cluster {#synchronization}

When working with Python Spark via DataSphere, there is no need to manually migrate the virtual environment. In a Yandex Data Processing cluster, you can change the basic composition of PyPI packages using a virtual environment:

1. Install the `catboost` library:

   ```python
   %pip install catboost
   ```

1. Once the installation is complete, from the top panel, select **Kernel** ⟶ **Restart kernel...**. If the installation completes without [errors](../troubleshooting/troubles-with-spark.md), the virtual environment will be automatically created and available in the Spark session using the `spark` variable.

To synchronize the environment, the Spark connector settings in the **S3 settings** section must specify the [static access key](../../iam/concepts/authorization/access-key.md) ID for the bucket and the [secret](secrets.md) containing the static access key.

{% note warning %}

Python environment synchronization is running in test mode. To enable environment synchronization, in the Spark connector settings, under **Spark settings**, specify the `.options` = `venv` parameter.

{% endnote %}

## Livy sessions {#livy-sessions}

For correct integration with DataSphere via Livy sessions, make sure the [image version](../../data-proc/concepts/environment.md) of the deployed Yandex Data Processing cluster is at least `2.0`, with `LIVY`, `SPARK`, `YARN`, and `HDFS` enabled.

{% note info %}

To get more than 100 MB of Yandex Data Processing cluster data, use an [S3 connector](../operations/data/s3-connectors.md).

{% endnote %}

### Computing sessions {#session}

In Yandex Data Processing clusters, your code is running in [sessions](https://livy.incubator.apache.org/docs/latest/rest-api.html#session). The session stores intermediate results until you delete it or shut down the cluster. Each cluster has a default session. The session ID is the same as the project ID.

Use the following commands to manage sessions:
* `%create_livy_session --cluster <cluster_name> --id <session_ID>`: Creating a session.
* `%delete_livy_session --cluster <cluster_name> --id <session_ID>`: Deleting a session.

For example, the command below creates a session named `ses1` in `my-new-cluster`, which allows each process to use a maximum of 4 CPUs per cluster and 4 GB of RAM (for more information, see [these Spark articles](https://spark.apache.org/docs/latest/configuration.html)):

```python
%create_livy_session --cluster my-new-cluster --id ses1 --conf spark.cores.max=4 --conf spark.executor.memory=4g
```

Dynamic resource allocation is enabled in sessions by default. To limit the session's resources, set `--conf spark.dynamicAllocation.enabled=false`.

### Livy session parameters {#parameters}

Complete list of parameters for the `%create_livy_session` command:

| Parameter                     | Type      | Description                                   |
|------------------------------|----------|--------------------------------------------|
| `--cluster`                  | `string` | Yandex Data Processing cluster ID or name |
| `--id`                       | `string` | Session ID, any string. If not set, the system will generate it automatically. |
| `--conf`                     | `string` | Spark configuration properties                |
| `--proxyUser`                | `string` | Yandex Data Processing cluster OS username that will be used to run the job. The default value is `spark`. |
| `--jars`                     | `string` | Java libraries to use in the session |
| `--files`                    | `string` | Files to use in the session           |
| `--pyFiles`                  | `string` | Python files to use in the session    |
| `--driverMemory`             | `string` | Driver memory                      |
| `--driverCores`              | `int`    | Number of driver cores               |
| `--executorMemory`           | `string` | Worker memory          |
| `--executorCores`            | `int`    | Number of worker cores      |
| `--numExecutors`             | `int`    | Number of workers          |
| `--archives`                 | `string` | Archives to use in the session          |
| `--queue`                    | `string` | YARN queue name                      |
| `--variables`                | `string` | Variables to use in the session      |
| `--return_variables`         | `string` | Variables to return by the session  |
| `--heartbeatTimeoutInSecond` | `int`    | Time to wait for heartbeat before the session ends     |
| `--ttl`                      | `string` | Inactive session timeout          |

Learn more about the Livy session parameters in these [official guides](https://livy.incubator.apache.org/docs/latest/rest-api.html).

### Yandex Data Processing session restrictions {#restrictions}

DataSphere uses system variables to run Yandex Data Processing clusters. Do not override the values of the following variables:
* `sc`
* `spark`
* `HiveContext`
* `StreamingContext`
* `SqlContext`

The parameters required to run Livy jobs override the following global Spark configurations:
* `spark.jars`
* `spark.submit.deployMode`
* `spark.yarn.dist.archives`
* `spark.submit.pyFiles`
* `spark.yarn.maxAppAttempts`
* `spark.yarn.submit.waitAppCompletion`

To specify additional libraries for a Spark session, use the `spark.driver.extraClassPath` and `spark.executor.extraClassPath` parameters and place the libraries on all the nodes when creating a Yandex Data Processing cluster with initialization scripts. Make sure the paths to the libraries you use are the same on all cluster nodes.

### Running Python code in a cluster {#run-code}

The code is run in the cells with the header:

```text
#!spark [--cluster <cluster>] [--session <session>] [--variables <input_variable>] [--return_variables <returned_variable>]
```

Where:

* `--cluster`: Yandex Data Processing cluster to perform computations on. This can be:
  * Name of the cluster created through the notebook interface.
  * HTTP link to the internal IP address of the `masternode` host, such as `http://10.0.0.8:8998/`.
* `--session`: Computing session ID. If this parameter is not specified, the default Yandex Data Processing cluster session is used.
* `--variables`: Variable imported to the DataSphere cluster from Yandex Data Processing. Supported types include `bool`, `int`, `float`, `str`, and `pandas.DataFrame` (converted to Spark DataFrame in a cluster).
* `--return_variables`: Variable to be exported from the Yandex Data Processing cluster to DataSphere. Supported type: `bool`, `int`, `float`, `str`, `pandas.DataFrame` (transformed Spark DataFrame).

### Example of using computing sessions with user-defined parameters {#example-custom-sessions}

To run computations in a session with defined settings, first create this session and then provide the code in the cell with the `#!spark` header:

1. Create a session and set its parameters:

   ```python
   %create_livy_session --cluster my-new-cluster --id ses1 --conf spark.cores.max=4 --conf spark.executor.memory=4g
   ```

1. In the next cell, run your computations:

   ```python
   #!spark --cluster my-new-cluster --session ses1
   import random

   def inside(p):
       x, y = random.random(), random.random()
       return x*x + y*y < 1

   NUM_SAMPLES = 1_000_000

   count = sc.parallelize(range(0, NUM_SAMPLES)) \
       .filter(inside).count()
   print("Pi is roughly %f" % (4.0 * count / NUM_SAMPLES))
   ```

1. If you no longer need the session, delete it:

   ```python
   %delete_livy_session --cluster my-new-cluster --id ses1
   ```

### Working with the Spark SQL library {#sql}

DataSphere supports using the Spark SQL library. For example, the query below will return all records in the `animals` table created in `cluster test-dataproc-cluster`:

```python
#!spark --cluster test-dataproc-cluster --return_variables df
df = spark.sql("SELECT * FROM animals;")
```

```python
df
```

For more information about SQL query syntax and working with the Spark SQL library, see [this Spark guide](https://spark.apache.org/docs/latest/sql-ref-syntax-qry-select.html).

## Useful links {#see-also}

* [Integration with Yandex Data Processing](../tutorials/data-processing-integration.md)
* [Spark connector](spark-connector.md)