[Yandex Cloud documentation](../../index.md) > [Yandex Managed Service for Apache Kafka®](../index.md) > [Tutorials](index.md) > Transferring data from Managed Service for Apache Kafka® into a Apache Iceberg™ table in Object Storage

# Transferring data from a Yandex Managed Service for Apache Kafka® topic into a Apache Iceberg™ table in Yandex Object Storage

# Transferring data from a Yandex Managed Service for Apache Kafka® topic into an Apache Iceberg™ table in Yandex Object Storage

You can set up data transfer from a [Managed Service for Apache Kafka® topic](../concepts/topics.md#topics) into an Apache Iceberg™ table. Use [Yandex Managed Service for Trino](../../managed-trino/index.md) to create a schema and a table. The data will be stored in an [Object Storage](../../storage/index.md) bucket, and the metadata, in [Apache Hive™ Metastore](../../metadata-hub/concepts/metastore.md). In Managed Service for Apache Kafka®, configure [Iceberg Sink connector](../concepts/connectors.md#iceberg-sink) to transfer data from a topic to the table.

You can also use a PySpark job to create an Apache Iceberg™ table via [Yandex Managed Service for Apache Spark™](../../managed-spark/index.md). For more information, see [this tutorial](../../managed-spark/tutorials/spark-simple-rw-job.md).

To set up data transfer:

1. [Set up your infrastructure](#infra).
1. [Create an Apache Iceberg™ table](#create-iceberg-table).
1. [Create a connector and test it](#connector-create).

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}

* Managed Service for Apache Kafka® cluster: use of computing resources allocated to hosts and storage size (see [Managed Service for Apache Kafka® pricing](../pricing.md)).
* Public IP addresses if public access is enabled for cluster hosts (see [Yandex Virtual Private Cloud pricing](../../vpc/pricing.md)).
* Apache Hive™ Metastore cluster: computing resources of the cluster components (see [Yandex MetaData Hub pricing](../../metadata-hub/pricing.md)).
* Managed Service for Trino cluster: computing resources of the cluster components and the amount of Yandex Cloud egress traffic (see [Managed Service for Trino pricing](../../managed-trino/pricing.md)).
* Object Storage bucket: use of storage, data operations (see [Object Storage pricing](../../storage/pricing.md)).
* NAT gateway: hourly use of the gateway and its outgoing traffic (see [Virtual Private Cloud pricing](../../vpc/pricing.md)).


## Set up your infrastructure {#infra}

1. [Create a service account](../../iam/operations/sa/create.md#create-sa) named `sa-metastore` and assign it the following roles:

    * [storage.editor](../../storage/security/index.md#storage-editor): For using the Object Storage bucket.
    * [managed-metastore.integrationProvider](../../metadata-hub/security/metastore-roles.md#managed-metastore-integrationProvider): For [communication between the Apache Hive™ Metastore cluster and Yandex Cloud services](../../metadata-hub/concepts/metastore-impersonation.md).

1. Create a service account named `sa-trino` and assign it the following roles:

    * [storage.editor](../../storage/security/index.md#storage-editor): For using the Object Storage bucket.
    * [managed-trino.integrationProvider](../../managed-trino/security.md#managed-trino-integrationProvider): For [communication between the Managed Service for Trino cluster and Yandex Cloud services](../../managed-trino/concepts/impersonation.md).

1. [Create an Object Storage bucket](../../storage/operations/buckets/create.md).
1. [Create a static access key](../../iam/operations/authentication/manage-access-keys.md#create-access-key) for the `sa-metastore` service account.

    Save the key ID and secret key, as you will need them when [creating a connector](#connector-create).

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

    This will automatically create three subnets in different availability zones.

1. [Set up a NAT gateway](../../vpc/operations/create-nat-gateway.md) for the `demo-network-ru-central1-a` subnet.

    A NAT gateway is required for communication between the Apache Hive™ Metastore cluster and Yandex Cloud services.

1. In `demo-network`, [create a security group](../../vpc/operations/security-group-create.md) named `metastore-sg` for the Apache Hive™ Metastore cluster and [add the rules](../../metadata-hub/operations/metastore/configure-security-group.md) required for the cluster to that group.

1. In `demo-network`, create a security group named `mkf-sg` for the Managed Service for Apache Kafka® cluster and add the following rules to that group:

    * Rule for incoming traffic allowing connections to the cluster over the internet:

      * **Port range**: `9091`.
      * **Protocol**: `TCP`.
      * **Source**: `Address range`.
      * **IPv4 CIDR**: `0.0.0.0/0`.

    * Rule for outgoing traffic allowing access to Apache Hive™ Metastore:

      * **Port range**: `9083`.
      * **Protocol**: `TCP`.
      * **Destination name**: `Address range`.
      * **IPv4 CIDR**: `0.0.0.0/0`.

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

    * **Service account**: `sa-metastore`.
    * **Bucket name**: Name of the bucket.
    * **Network**: `demo-network`.
    * **Subnet**: `demo-network-ru-central1-a`.
    * **Security groups**: `metastore-sg`.

1. [Create a Managed Service for Trino cluster](../../managed-trino/operations/cluster-create.md) with the following settings:

    * **Network**: `demo-network`.
    * **Service account**: `sa-trino`.
    * Catalog properties:

      * **Catalog name**: `iceberg`.
      * **Connector type**: `Iceberg`.
      * **Metastore type**: `Hive Metastore`.
      * **URI**: `thrift://<Metastore_cluster_IP_address>:9083`.

        You can get the Apache Hive™ Metastore cluster IP address along with the [cluster info](../../metadata-hub/operations/metastore/cluster-list.md#get-cluster).

      * **File storage**: `Yandex Object Storage`.

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

    * **Network**: `demo-network`.
    * **Security groups**: `mkf-sg`.
    * **Public access**: Enabled.

      {% note info %}
      
      Public access to cluster hosts is required if you plan to connect to the cluster via the internet. This connection option is simpler and is recommended for the purposes of this guide. You can connect to non-public hosts as well but only from Yandex Cloud virtual machines located in the same cloud network as the cluster.
      
      {% endnote %}

1. In your Managed Service for Apache Kafka® cluster, [create the following topics](../operations/cluster-topics.md#create-topic):

    * `iceberg_control_topic`: For managing the connector.
    * `my_topic`: For messaging.

1. In your Managed Service for Apache Kafka® cluster, [create a user](../operations/cluster-accounts.md#create-account) named `kafka-producer` with the [producer](../concepts/producers-consumers.md) role and grant them access to `my_topic`.

    This user is used to publish messages to `my_topic`.


## Create an Apache Iceberg™ table {#create-iceberg-table}

1. [Connect](../../managed-trino/operations/connect.md) to your Managed Service for Trino cluster.
1. Run the following SQL queries:

    1. Create a schema:

        ```sql
        CREATE SCHEMA iceberg.myschema
        WITH (
          location = 's3a://<bucket_name>/iceberg/warehouse/myschema'
        );
        ```

    1. Create a table:

        ```sql
        CREATE TABLE iceberg.myschema.mytable (
          id BIGINT,
          name VARCHAR,
          created_at VARCHAR
        )
        WITH (format = 'PARQUET');
        ```

1. Make sure you have successfully created the `iceberg/warehouse/myschema/mytable-*/metadata/` [folder structure](../../storage/operations/objects/list.md), where `*` is the system component of the table name.


## Create a connector and test it {#connector-create}

1. In Managed Service for Apache Kafka®, [create a connector](../operations/cluster-connector.md#create) of the `Iceberg Sink` type with the following settings:

    * **Control topic**: `iceberg_control_topic`.
    * **Topics**: `my_topic`.
    * **Tables**: `myschema.mytable`.
    * **Catalog URI**: `thrift://<Metastore_cluster_IP_address>:9083`.

      You can get the Apache Hive™ Metastore cluster IP address along with the [cluster info](../../metadata-hub/operations/metastore/cluster-list.md#get-cluster).

    * **Warehouse**: `s3a://<bucket_name>/iceberg/warehouse`.
    * **Endpoint**: `storage.yandexcloud.net`.
    * **Access key ID**, **Secret access key**: Static key data [you got earlier](#infra).
    * **Commit interval, ms**: `5000` milliseconds.
    * **Additional properties**:

      * `key.converter`: `org.apache.kafka.connect.json.JsonConverter`
      * `key.converter.schemas.enable`: `false`
      * `value.converter`: `org.apache.kafka.connect.json.JsonConverter`
      * `value.converter.schemas.enable`: `false`

1. Test the connector:

    1. [Install an SSL certificate](../operations/connect/index.md#get-ssl-cert).
    1. [Install kcat (kafkacat)](../operations/connect/clients.md#bash-zsh).
    1. Send a message to `my_topic`:

        ```bash
        echo '{"id":1,"name":"Alice","created_at":"2024-01-15T10:30:00"}' | kcat -P \
            -b <broker_FQDN>:9091 \
            -t my_topic \
            -X security.protocol=SASL_SSL \
            -X sasl.mechanism=SCRAM-SHA-512 \
            -X sasl.username="kafka-producer" \
            -X sasl.password="<password>" \
            -X ssl.ca.location=/usr/local/share/ca-certificates/Yandex/YandexInternalRootCA.crt
        ```

        For more information on getting a broker host FQDN, see [Getting FQDNs of Apache Kafka® hosts](../operations/connect/index.md#get-fqdn).

1. After committing, verify that the `data` folder containing data in `PARQUET` format has been created under `iceberg/warehouse/myschema/mytable-*/` in the bucket.

    To view the written data, run the following SQL query in Managed Service for Trino:

    ```sql
    SELECT * FROM iceberg.myschema.mytable;
    ```


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

Some resources are not free of charge. Delete the resources you no longer need, so you will not be charged for them:

1. [Managed Service for Apache Kafka® cluster](../operations/cluster-delete.md).
1. [Apache Hive™ Metastore cluster](../../metadata-hub/operations/metastore/cluster-delete.md).
1. [Managed Service for Trino cluster](../../managed-trino/operations/cluster-delete.md).
1. [Object Storage bucket](../../storage/operations/buckets/delete.md). Before deleting the bucket, [delete all its objects](../../storage/operations/objects/delete.md).
1. [NAT gateway](../../vpc/operations/delete-nat-gateway.md).