Data Lake Formation:EMR Serverless Spark Mengakses Katalog DLF Menggunakan Iceberg REST

Prasyarat

Anda telah membuat ruang kerja Serverless Spark di wilayah yang sama dengan instans DLF Anda. Untuk informasi selengkapnya, lihat Buat ruang kerja.

Batasan

Tipe tugas berikut didukung:

• Sesi SQL: Kelola sesi SQL.

• Spark Thrift Server: Kelola sesi Spark Thrift Server.

• Pekerjaan Batch: Kembangkan pekerjaan batch.

Langkah 1: Berikan izin katalog

2. Pada halaman Catalogs, klik nama katalog untuk membuka halaman detailnya.

3. Untuk memberikan izin ke seluruh katalog, klik tab Permissions. Atau, Anda dapat menavigasi ke database atau tabel tertentu dan mengklik tab Permissions-nya untuk memberikan akses.

4. Pada halaman otorisasi, konfigurasikan pengaturan berikut lalu klik OK.

   • User/Role: Pilih RAM User/RAM Role.

   • Select Authorization Object: Pilih AliyunECSInstanceForEMRRole dari daftar dropdown.

     Catatan

     Jika AliyunECSInstanceForEMRRole tidak muncul dalam daftar dropdown, buka halaman manajemen pengguna lalu klik Sync.

   • Preset Permission Type: Pilih izin baca secara manual atau gunakan peran yang telah ditentukan sebelumnya, seperti Data Reader atau Data Editor.

Langkah 2: Baca dan tulis data

1. Anda dapat menghubungkan ke katalog tersebut.

   Anda dapat membuat sesi SQL. Untuk informasi selengkapnya, lihat Kelola sesi SQL. Anda dapat menggunakan versi engine esr-4.7.0, esr-3.6.0, atau yang lebih baru.

   Gunakan katalog data

   Jika Anda menggunakan katalog data, Anda tidak perlu mengonfigurasi parameter dalam sesi. Buka halaman Data Catalog, lalu klik Add data catalog. Setelah itu, Anda dapat langsung memilih katalog data tersebut dalam pengembangan SparkSQL.

   Gunakan katalog kustom

   Pada bagian Spark Configuration di Custom Configuration, tambahkan konfigurasi berikut.

   Penting

   Pada contoh konfigurasi berikut, iceberg_catalog adalah nama katalog kustom. Katalog ini mendaftarkan layanan manajemen tabel Iceberg di Spark berdasarkan katalog Iceberg REST dan terhubung ke Alibaba Cloud DLF melalui REST API. Anda dapat mengubah nama katalog dan parameter terkait sesuai kebutuhan.

   • ${regionID}: Ganti dengan ID wilayah yang sebenarnya, misalnya cn-hangzhou. Untuk informasi selengkapnya, lihat Endpoint.

   • ${catalogName}: Ganti dengan nama katalog DLF Anda.

   • ${access_key_id} dan ${access_key_secret}: Ganti dengan ID AccessKey dan rahasia Akun Alibaba Cloud Anda.

   # Aktifkan ekstensi Iceberg Spark
   spark.sql.extensions                                org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
   # Daftarkan katalog iceberg_catalog
   spark.sql.catalog.iceberg_catalog                   org.apache.iceberg.spark.SparkCatalog
   # Implementasi katalog dasar adalah katalog Iceberg REST
   spark.sql.catalog.iceberg_catalog.catalog-impl      org.apache.iceberg.rest.RESTCatalog
   # Titik akhir REST API layanan DLF Iceberg.
   spark.sql.catalog.iceberg_catalog.uri               http://${regionID}-vpc.dlf.aliyuncs.com/iceberg
   # Tentukan nama katalog DLF Anda.
   spark.sql.catalog.iceberg_catalog.warehouse         ${catalogName}
   # Gunakan implementasi DLF FileIO kustom
   spark.sql.catalog.iceberg_catalog.io-impl           org.apache.iceberg.rest.DlfFileIO
   # Aktifkan otentikasi signature SigV4
   spark.sql.catalog.iceberg_catalog.rest.auth.type    sigv4
   spark.sql.catalog.iceberg_catalog.rest.auth.sigv4.delegate-auth-type  none
   spark.sql.catalog.iceberg_catalog.rest.signing-region  ${regionID}
   spark.sql.catalog.iceberg_catalog.rest.signing-name  DlfNext
   # Kredensial akses
   spark.sql.catalog.iceberg_catalog.rest.access-key-id ${access_key_id}
   spark.sql.catalog.iceberg_catalog.rest.secret-access-key ${access_key_secret}

2. Baca dan tulis data.

   Untuk contoh lengkap pengembangan pekerjaan SQL, lihat Memulai pengembangan SparkSQL.

   Catatan

   Jika tidak ada database yang ditentukan, tabel akan dibuat di database default katalog tersebut. Anda juga dapat membuat dan menentukan database lain.

   -- Buat database
   CREATE DATABASE IF NOT EXISTS db;
   
   -- Buat tabel non-partisi
   CREATE TABLE iceberg_catalog.db.tbl (
       id BIGINT NOT NULL COMMENT 'unique id',
       data STRING
   )
   USING iceberg;
   
   -- Masukkan data ke dalam tabel non-partisi
   INSERT INTO iceberg_catalog.db.tbl VALUES
   (1, 'Alice'),
   (2, 'Bob'),
   (3, 'Charlie');
   
   -- Kueri seluruh tabel
   SELECT * FROM iceberg_catalog.db.tbl;
   
   -- Kueri bersyarat
   SELECT * FROM iceberg_catalog.db.tbl WHERE id = 2;
   
   -- Perbarui data
   UPDATE iceberg_catalog.db.tbl SET data = 'David' WHERE id = 3;
   
   -- Kueri ulang tabel untuk mengonfirmasi pembaruan
   SELECT * FROM iceberg_catalog.db.tbl WHERE id = 3;
   
   -- Hapus data
   DELETE FROM iceberg_catalog.db.tbl WHERE id = 1;
   
   -- Kueri ulang tabel untuk mengonfirmasi penghapusan
   SELECT * FROM iceberg_catalog.db.tbl;
   
   -- Buat tabel partisi
   CREATE TABLE iceberg_catalog.db.part_tbl (
       id BIGINT,
       data STRING,
       category STRING,
       ts TIMESTAMP
   )
   USING iceberg
   PARTITIONED BY (category);
   
   -- Masukkan data
   INSERT INTO iceberg_catalog.db.part_tbl VALUES
   (100, 'Data1', 'A', to_timestamp('2025-01-01 12:00:00')),
   (200, 'Data2', 'B', to_timestamp('2025-01-02 14:00:00')),
   (300, 'Data3', 'A', to_timestamp('2025-01-01 15:00:00')),
   (400, 'Data4', 'C', to_timestamp('2025-01-03 10:00:00'));
   
   
   -- Kueri seluruh tabel
   SELECT * FROM iceberg_catalog.db.part_tbl;
   
   -- Kueri bersyarat
   SELECT * FROM iceberg_catalog.db.part_tbl WHERE bucket(16, id) = 0;
   
   -- Kueri bersyarat
   SELECT * FROM iceberg_catalog.db.part_tbl WHERE days(ts) = '2025-01-01';
   
   -- Kueri bersyarat
   SELECT * FROM iceberg_catalog.db.part_tbl WHERE category = 'A';
   
   -- Kueri bersyarat gabungan (bucket + hari + kategori)
   SELECT * FROM iceberg_catalog.db.part_tbl 
   WHERE bucket(16, id) = 0 
     AND days(ts) = '2025-01-01'
     AND category = 'A';
   
   -- Agregasi dan hitung jumlah entri data untuk setiap kategori
   SELECT category, COUNT(*) AS count 
   FROM iceberg_catalog.db.part_tbl 
   GROUP BY category;
   
   -- Hapus database. Pastikan semua tabel di dalamnya kosong.
   -- DROP DATABASE iceberg_catalog.db;