[Yandex Cloud documentation](../../index.md) > [Yandex Managed Service for Apache Spark™](../index.md) > [Tutorials](index.md) > Working with an Apache Iceberg™ table from a PySpark job

# Working with an Apache Iceberg™ table from a PySpark job

# Working with an Apache Iceberg™ table from a PySpark job


Yandex Managed Service for Apache Spark™ features the integrated Apache Iceberg™ library. To use Apache Iceberg™ in a PySpark job, connect the Apache Hive™ Metastore server to the Yandex Managed Service for Apache Spark™ cluster.

The Apache Iceberg™ format supports:
* Data versioning and snapshot storage.
* ACID transactions supporting `UPDATE`, `DELETE`, and `MERGE` operations, as well as the evolution of tables and the partitioning method.
* Scalability while maintaining high operational performance.

This tutorial demonstrates how to use the Apache Iceberg™ table format to create and read metadata snapshots. The Apache Iceberg™ table is created in the S3 Object Storage. The Yandex Managed Service for Apache Spark™ cluster where the PySpark job runs uses the Apache Hive™ Metastore global catalog. For more information on connecting to the global catalog, see [this guide](metastore-and-spark.md).

To implement the above example:

1. [Set up your infrastructure](#infra).
1. [Prepare and run a PySpark job](#prepare-job).
1. [Check the result](#check-out).

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


## Required paid resources {#paid-resources}

* Object Storage buckets: use of storage and data operations (see [Object Storage pricing](../../storage/pricing.md)).
* Yandex Cloud Logging: amount of written data and its retention time (see [Cloud Logging pricing](../../logging/pricing.md)).
* Yandex Managed Service for Apache Spark™ cluster: computing resources of cluster components (see [Yandex Managed Service for Apache Spark™ pricing](../pricing.md)).
* Apache Hive™ Metastore cluster: computing resources of cluster components (see [Yandex MetaData Hub pricing](../../metadata-hub/pricing.md)).


## Set up your infrastructure {#infra}

Set up your infrastructure:

1. [Create a service account](../../iam/operations/sa/create.md) named `spark-agent` and assign it the [managed-spark.integrationProvider](../../iam/roles-reference.md#managed-spark-integrationProvider) role.

1. [Create a service account](../../iam/operations/sa/create.md) named `metastore-agent` and assign it the [managed-metastore.integrationProvider](../../iam/roles-reference.md#managed-metastore-integrationProvider) role to enable your Apache Hive™ Metastore cluster to [interact with other resources](../../metadata-hub/concepts/metastore-impersonation.md).

1. [Create buckets](../../storage/operations/buckets/create.md):

    * One for the PySpark job source code.
    * One for output data.

1. [Grant permissions](../../storage/operations/buckets/edit-acl.md) to the `spark-agent` service account for the created buckets:

    * Bucket for the PySpark job source code: `READ`.
    * Bucket for output data: `READ and WRITE`.

1. [Grant](../../storage/operations/buckets/edit-acl.md) `metastore-agent` the `READ and WRITE` permissions for the output bucket.

1. [Create a cloud network](../../vpc/operations/network-create.md) named `integration-network`.

    This will automatically create three subnets in different availability zones.

1. For the Yandex Managed Service for Apache Spark™ cluster, [create a security group](../../vpc/operations/security-group-create.md) named `spark-sg` in `integration-network`. Add the following rule to it:

    * For outgoing traffic, to allow Yandex Managed Service for Apache Spark™ cluster connections to Apache Hive™ Metastore:

        * Port range: `9083`
        * Protocol: `Any`
        * Destination: `CIDR`
        * CIDR blocks: `0.0.0.0/0`

1. For the Apache Hive™ Metastore cluster, [create a security group](../../vpc/operations/security-group-create.md) named `metastore-sg` in `integration-network`. Add the following rules to it:

    * For incoming client traffic:

        * Port range: `30000-32767`
        * Protocol: `Any`
        * Source: `CIDR`
        * CIDR blocks: `0.0.0.0/0`

    * For incoming load balancer traffic:

        * Port range: `10256`
        * Protocol: `Any`
        * Source: `Load balancer health checks`

1. [Create a Apache Hive™ Metastore cluster](../../metadata-hub/operations/metastore/cluster-create.md) with the following parameters:

    * **Service account**: `metastore-agent`.
    * **Version**: `3.1`.
    * **Network**: `integration-network`.
    * **Subnet**: `integration-network-ru-central1-a`.
    * **Security groups**: `metastore-sg`.

1. [Create a Yandex Managed Service for Apache Spark™ cluster](../operations/cluster-create.md) with the following parameters:

    * **Service account**: `spark-agent`.
    * **Network**: `integration-network`.
    * **Subnet**: `integration-network-ru-central1-a`.
    * **Security groups**: `spark-sg`.
    * **Metastore**: Apache Hive™ Metastore cluster you created earlier.

## Prepare and run a PySpark job {#prepare-job}

For a PySpark job, we will use a Python script that:

1. Creates a database and table in Apache Iceberg™ format in the bucket.
1. Writes 10 rows of data to the table.
1. Stores the ID of the current table snapshot.
1. Writes 10 more rows of data to the table.
1. Displays the number of rows in the current table state.
1. Displays the number of rows in the table state at the time of the snapshot. 

Prepare a script file:

1. Create the `ice_min_demo.py` file and paste the following code to it:

   {% cut "ice_min_demo.py" %}

   ```python
   import random
   from pyspark.sql import SparkSession

   spark = (
      SparkSession.builder
      .appName("ice_min_demo")
      .enableHiveSupport()
      .getOrCreate()
   )

   # Creating a database and table in Apache Iceberg™ format
   # Apache Hive™ Metastore captures metadata, allowing you to access the table by the `db.tbl` name
   # from any Spark apps associated with this Apache Hive™ Metastore cluster.
   db, tbl = "demo_db", "demo_events"
   spark.sql(f"CREATE DATABASE IF NOT EXISTS {db}")
   spark.sql(f"""
   CREATE TABLE IF NOT EXISTS {db}.{tbl} (
      id BIGINT,
      value DOUBLE
   ) USING iceberg
   """)

   # Writing the first piece of data to the table
   df1 = spark.createDataFrame([(i, random.random()) for i in range(10)], ["id","value"])
   df1.writeTo(f"{db}.{tbl}").append()

   # Fetching the ID of the current snapshot from the `.snapshots` housekeeping table
   snap_before = spark.sql(f"SELECT max(snapshot_id) AS s FROM {db}.{tbl}.snapshots").collect()[0][0]

   # Writing the second piece of data to the table
   df2 = spark.createDataFrame([(i, random.random()) for i in range(10, 20)], ["id","value"])
   df2.writeTo(f"{db}.{tbl}").append()

   # Counting and displaying the number of rows in the table's current (20) and previous (10) state
   cnt_now = spark.table(f"{db}.{tbl}").count()
   cnt_past = spark.sql(f"SELECT COUNT(*) FROM {db}.{tbl} VERSION AS OF {snap_before}").collect()[0][0]
   print(f"now_count: {cnt_now} | past_count: {cnt_past}", flush=True)

   spark.stop()
   ```

   {% endcut %}

1. In the source code bucket, create a folder named `scripts` and [upload](../../storage/operations/objects/upload.md#simple) the `ice_min_demo.py` file to this folder.
1. [Create a job](../operations/jobs-pyspark.md) with the following settings:
    * **Job type**: **PySpark**.
    * **Main python file**: `s3a://<source_code_bucket>/scripts/ice_min_demo.py`.
    * **Arguments**: `spark.sql.warehouse.dir` – `s3a://<output_data_bucket>/warehouse/`.

## Check the result {#check-out}

1. Wait for the PySpark job you created to change its status to **Done**.
1. Open the [job execution logs](../operations/jobs-pyspark.md#get-logs).
1. In the logs, navigate to this string: `now_count: 20 | past_count: 10`.
1. Make sure the `warehouse/demo_db` folder appears in your output data bucket. The data from the new database, `demo_db`, is now stored in the Object Storage bucket, and the DB metadata, in the Apache Hive™ Metastore cluster.
 
## 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. [Object Storage buckets](../../storage/operations/buckets/delete.md). Before deleting your buckets, make sure to [have deleted all objects from those buckets](../../storage/operations/objects/delete.md).
1. [Apache Hive™ Metastore cluster](../../metadata-hub/operations/metastore/cluster-delete.md).
1. [Yandex Managed Service for Apache Spark™ cluster](../operations/cluster-delete.md).