Gunakan tabel Paimon di EMR Serverless Spark - E-MapReduce

Apache Paimon adalah format data lake terpadu untuk pemrosesan streaming dan batch yang mendukung penulisan throughput tinggi serta kueri latensi rendah. Topik ini menjelaskan cara membaca dan menulis tabel Paimon di EMR Serverless Spark.

Prasyarat

Ruang kerja telah dibuat. Buat ruang kerja.

Prosedur

Langkah 1: Buat sesi SQL

Buka halaman Sessions.
1. Login ke Konsol EMR.
2. Pada panel navigasi kiri, pilih EMR Serverless > Spark.
3. Pada halaman Spark, klik nama ruang kerja yang dituju.
4. Pada halaman EMR Serverless Spark, klik Sessions di panel navigasi kiri.
Pada halaman SQL Sessions, klik Create SQL Session.

Pada halaman Create SQL Session, konfigurasikan parameter pada bagian Spark Configuration, lalu klik Create. Kelola sesi SQL.

Spark membaca dan menulis tabel Paimon melalui katalog. Pilih jenis katalog berdasarkan skenario Anda. Kelola katalog data.

Katalog data

Saat menggunakan katalog data, tidak diperlukan parameter sesi. Klik Catalogs pada halaman Add Catalog, lalu pilih katalog data di lingkungan pengembangan Spark SQL Anda.

Catatan

Disarankan menggunakan versi mesin EMR esr-4.3.0 atau lebih baru, esr-3.3.0 atau lebih baru, atau esr-2.7.0 atau lebih baru.

Katalog kustom

Data Lake Formation (DLF)

spark.sql.catalog.<catalogName>                                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore                       rest
spark.sql.catalog.<catalogName>.uri                             http://cn-hangzhou-vpc.dlf.aliyuncs.com
spark.sql.catalog.<catalogName>.warehouse                       <catalog_name>
spark.sql.catalog.<catalogName>.token.provider                  dlf
spark.sql.catalog.<catalogName>.dlf.access-key-id               <access_key_id>
spark.sql.catalog.<catalogName>.dlf.access-key-secret           <access_key_secret>

Parameter:

Parameter	Deskripsi	Nilai contoh
`spark.sql.catalog.<catalogName>`	Implementasi katalog.	Nilai tetap: `org.apache.paimon.spark.SparkCatalog`
`spark.sql.catalog.<catalogName>.metastore`	Metode penyimpanan metadata. Atur ke `rest` untuk menggunakan API REST DLF.	Nilai tetap: `rest`
`spark.sql.catalog.<catalogName>.uri`	URI DLF. Format: `http://<endpoint>-vpc.dlf.aliyuncs.com`.	`http://cn-hangzhou-vpc.dlf.aliyuncs.com`
`spark.sql.catalog.<catalogName>.warehouse`	Jalur gudang. Untuk DLF, atur nilai ini ke nama katalog.	`<catalog_name>`
`spark.sql.catalog.<catalogName>.token.provider`	Penyedia otentikasi. Atur ke `dlf` untuk DLF.	Nilai tetap: `dlf`
`spark.sql.catalog.<catalogName>.dlf.access-key-id`	ID AccessKey Akun Alibaba Cloud atau Pengguna RAM Anda.	`<access_key_id>`
`spark.sql.catalog.<catalogName>.dlf.access-key-secret`	Rahasia AccessKey Akun Alibaba Cloud atau Pengguna RAM Anda.	`<access_key_secret>`

DLF-Legacy

Metadata disimpan di DLF 1.0.

spark.sql.catalog.<catalogName>                          org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore                dlf
spark.sql.catalog.<catalogName>.dlf.catalog.id           <catalog_name>
spark.sql.catalog.<catalogName>.dlf.catalog.endpoint     dlf-vpc.cn-hangzhou.aliyuncs.com

Parameter:

Parameter	Deskripsi	Nilai contoh
`spark.sql.catalog.<catalogName>`	Implementasi katalog.	Nilai tetap: `org.apache.paimon.spark.SparkCatalog`
`spark.sql.catalog.<catalogName>.metastore`	Metode penyimpanan metadata. Atur ke `dlf` untuk menggunakan DLF sebagai metastore.	Nilai tetap: `dlf`
`spark.sql.catalog.<catalogName>.dlf.catalog.id`	Nama katalog di DLF.	`<catalog_name>`
`spark.sql.catalog.<catalogName>.dlf.catalog.endpoint`	Titik akhir DLF. Pilih titik akhir DLF berdasarkan Wilayah Anda.	`dlf-vpc.cn-hangzhou.aliyuncs.com`

Hive Metastore

Metadata disimpan di Hive Metastore tertentu.

spark.sql.catalog.<catalogName>                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore       hive
spark.sql.catalog.<catalogName>.uri             thrift://<yourHMSUri>:<port>

Parameter:

Parameter	Deskripsi	Nilai contoh
`spark.sql.catalog.<catalogName>`	Implementasi katalog.	Nilai tetap: `org.apache.paimon.spark.SparkCatalog`
`spark.sql.catalog.<catalogName>.metastore`	Jenis metastore. Atur ke `hive` untuk menggunakan Hive Metastore.	Nilai tetap: `hive`
`spark.sql.catalog.<catalogName>.uri`	URI Hive Metastore. Format: `thrift://<alamat IP Hive metastore>:9083`. `<alamat IP Hive metastore>` adalah alamat IP internal layanan Hive Metastore. Untuk menggunakan metastore eksternal, lihat Hubungkan ke Layanan Hive Metastore Eksternal.	`thrift://192.168..:9083`

sistem file

Metadata disimpan di sistem file.

spark.sql.catalog.<catalogName>                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore       filesystem
spark.sql.catalog.<catalogName>.warehouse       oss://<yourBucketName>/warehouse

Parameter:

Parameter	Deskripsi	Nilai contoh
`spark.sql.catalog.<catalogName>`	Implementasi katalog.	Nilai tetap: `org.apache.paimon.spark.SparkCatalog`
`spark.sql.catalog.<catalogName>.metastore`	Jenis metastore. Atur ke `filesystem` untuk menggunakan sistem file sebagai metastore.	Nilai tetap: `filesystem`
`spark.sql.catalog.<catalogName>.warehouse`	Jalur gudang. Dalam contoh ini, `<yourBucketName>` adalah nama bucket OSS.	`oss://my-bucket/warehouse`

Anda dapat mengonfigurasi beberapa katalog secara bersamaan, seperti DLF, DLF 1.0, dan Hive. Contoh:

# Konfigurasi katalog DLF
spark.sql.catalog.<catalogName>                                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore                       rest
spark.sql.catalog.<catalogName>.uri                             http://cn-hangzhou-vpc.dlf.aliyuncs.com
spark.sql.catalog.<catalogName>.warehouse                       <catalog_name>
spark.sql.catalog.<catalogName>.token.provider                  dlf
spark.sql.catalog.<catalogName>.dlf.access-key-id               <access_key_id>
spark.sql.catalog.<catalogName>.dlf.access-key-secret           <access_key_secret>

# Konfigurasi katalog DLF 1.0
spark.sql.catalog.<catalogName>                                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore                       dlf
spark.sql.catalog.<catalogName>.dlf.catalog.id                  <catalog_name>
spark.sql.catalog.<catalogName>.dlf.catalog.endpoint            dlf-vpc.cn-hangzhou.aliyuncs.com


# Konfigurasi katalog hive1
spark.sql.catalog.<catalogName>                                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore                       hive
spark.sql.catalog.<catalogName>.uri                             thrift://<yourHMSUri-1>:<port>

# Konfigurasi katalog hive2
spark.sql.catalog.<catalogName>                                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore                       hive
spark.sql.catalog.<catalogName>.uri                             thrift://<yourHMSUri-2>:<port>

Langkah 2: Baca dan tulis tabel Paimon

Buka halaman pengembangan SQL.

Pada halaman EMR Serverless Spark, klik Data Development di panel navigasi kiri.
Pada tab Development, klik ikon .
Pada kotak dialog Create, masukkan nama seperti users_task, pertahankan tipe default sebagai SparkSQL, lalu klik OK.

Salin kode berikut ke tab Spark SQL baru (users_task).

Katalog Paimon

-- Buat database.
CREATE DATABASE IF NOT EXISTS paimon.ss_paimon_db;             

-- Buat tabel Paimon.
CREATE TABLE paimon.ss_paimon_db.paimon_tbl (id INT, name STRING) USING paimon;

-- Tulis data ke tabel Paimon.
INSERT INTO paimon.ss_paimon_db.paimon_tbl VALUES (1, "a"), (2, "b"), (3, "c");

-- Kueri hasil tulis dari tabel Paimon.
SELECT * FROM paimon.ss_paimon_db.paimon_tbl ORDER BY id;

-- Hapus database.
DROP DATABASE paimon.ss_paimon_db CASCADE;

Pilih database dari daftar drop-down database dan sesi SQL yang telah Anda buat dari daftar drop-down sesi.
Klik Run untuk menjalankan pekerjaan. Pekerjaan akan menghasilkan output berikut.

FAQ

Mengapa saya menerima error saat menjalankan pernyataan DELETE, UPDATE, atau MERGE pada tabel?

Gejala: Menjalankan pernyataan DELETE, UPDATE, atau MERGE menghasilkan error seperti berikut:

Caused by: org.apache.spark.sql.AnalysisException: Table does not support deletes/updates/merge: <tableName>.
    at org.apache.spark.sql.errors.QueryCompilationErrors$.tableDoesNotSupportError(QueryCompilationErrors.scala:1391)

Penyebab: Format penyimpanan tabel tidak mendukung pembaruan tingkat baris, atau konfigurasi Spark yang diperlukan tidak tersedia.
Solusi:
1. Periksa jenis tabel.
  
  Jalankan perintah berikut untuk memeriksa apakah tabel tersebut merupakan tabel Paimon:
```
SHOW CREATE TABLE <tableName>;
```
  Jika output berisi USING PAIMON, maka tabel tersebut adalah tabel Paimon. Jika output menunjukkan format penyimpanan lain (seperti USING hive), pastikan format tersebut mendukung pembaruan tingkat baris.
2. Periksa konfigurasi Spark.
  
  Jika tabel tersebut adalah tabel Paimon, periksa bagian Spark Configuration dan pastikan konfigurasi berikut ada:
```
spark.sql.extensions org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions
```
  Jika tidak ada, tambahkan ke Spark Configuration.

Dokumentasi terkait

Untuk contoh lengkap proses pengembangan dan orkestrasi task SQL, lihat Panduan Cepat Pengembangan SparkSQL.
Untuk detail penggunaan dan konfigurasi Paimon, lihat dokumentasi resmi Paimon.
Untuk menggunakan layanan metastore eksternal, lihat Hubungkan ke Layanan Hive Metastore Eksternal.