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

You can set up data transfer from a Managed Service for Apache Kafka® topic into an Apache Iceberg™ table. Use Yandex Managed Service for Trino to create a schema and a table. The data will be stored in an Object Storage bucket, and the metadata, in Apache Hive™ Metastore. In Managed Service for Apache Kafka®, configure Iceberg Sink connector 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™. For more information, see this tutorial.

To set up data transfer:

1. Set up your infrastructure.
2. Create an Apache Iceberg™ table.
3. Create a connector and test it.

If you no longer need the resources you created, delete them.

Getting startedGetting started

Sign up for Yandex Cloud and create a billing account:

1. Navigate to the management console and log in to Yandex Cloud or create a new account.
2. On the Yandex Cloud Billing page, make sure you have a billing account linked and it has the ACTIVE or TRIAL_ACTIVE status. If you do not have a billing account, create one and link a cloud to it.

If you have an active billing account, you can create or select a folder for your infrastructure on the cloud page.

Learn more about clouds and folders here.

Required paid resourcesRequired 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).
• Public IP addresses if public access is enabled for cluster hosts (see Yandex Virtual Private Cloud pricing).
• Apache Hive™ Metastore cluster: computing resources of the cluster components (see Yandex MetaData Hub pricing).
• 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).
• Object Storage bucket: use of storage, data operations (see Object Storage pricing).
• NAT gateway: hourly use of the gateway and its outgoing traffic (see Virtual Private Cloud pricing).

Set up your infrastructureSet up your infrastructure

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

   • storage.editor: For using the Object Storage bucket.
   • managed-metastore.integrationProvider: For communication between the Apache Hive™ Metastore cluster and Yandex Cloud services.

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

   • storage.editor: For using the Object Storage bucket.
   • managed-trino.integrationProvider: For communication between the Managed Service for Trino cluster and Yandex Cloud services.

3. Create an Object Storage bucket.

4. Create a static access key for the sa-metastore service account.

   Save the key ID and secret key, as you will need them when creating a connector.

5. Create a cloud network named demo-network.

   This will automatically create three subnets in different availability zones.

6. Set up a NAT gateway 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.

7. In demo-network, create a security group named metastore-sg for the Apache Hive™ Metastore cluster and add the rules required for the cluster to that group.

8. 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.

9. Create a Apache Hive™ Metastore cluster 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.

10. Create a Managed Service for Trino cluster 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.

      • File storage: Yandex Object Storage.

11. Create a Managed Service for Apache Kafka® cluster with the following settings:

    • Network: demo-network.

    • Security groups: mkf-sg.

    • Public access: Enabled.

      Note

      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.

12. In your Managed Service for Apache Kafka® cluster, create the following topics:

    • iceberg_control_topic: For managing the connector.
    • my_topic: For messaging.

13. In your Managed Service for Apache Kafka® cluster, create a user named kafka-producer with the producer role and grant them access to my_topic.

    This user is used to publish messages to my_topic.

Create an Apache Iceberg™ tableCreate an Apache Iceberg™ table

1. Connect to your Managed Service for Trino cluster.

2. Run the following SQL queries:

   1. Create a schema:

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

   2. Create a table:

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

3. Make sure you have successfully created the iceberg/warehouse/myschema/mytable-*/metadata/ folder structure, where * is the system component of the table name.

Create a connector and test itCreate a connector and test it

1. In Managed Service for Apache Kafka®, create a connector 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.

   • Warehouse: s3a://<bucket_name>/iceberg/warehouse.

   • Endpoint: storage.yandexcloud.net.

   • Access key ID, Secret access key: Static key data you got earlier.

   • 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

2. Test the connector:

   1. Install an SSL certificate.

   2. Install kcat (kafkacat).

   3. Send a message to my_topic:

      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.

3. 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:

   SELECT * FROM iceberg.myschema.mytable;
   

Delete the resources you createdDelete the resources you created

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.
2. Apache Hive™ Metastore cluster.
3. Managed Service for Trino cluster.
4. Object Storage bucket. Before deleting the bucket, delete all its objects.
5. NAT gateway.