Baca data dari tabel Apache Paimon dengan membuat tabel eksternal dengan Flink-MaxCompute-Alibaba Cloud

Kueri data Apache Paimon yang disimpan di Object Storage Service (OSS) dari MaxCompute dengan memetakan tabel eksternal ke direktori tabel Paimon. Topik ini menjelaskan alur kerja end-to-end: tulis data Paimon ke OSS menggunakan Realtime Compute for Apache Flink, lalu kueri data tersebut dari MaxCompute.

Informasi latar belakang

Apache Paimon adalah format penyimpanan lake terintegrasi untuk pemrosesan streaming dan batch yang mendukung penulisan throughput tinggi serta kueri latensi rendah. Mesin komputasi umum seperti Spark, Hive, dan Trino dari Alibaba Cloud Realtime Compute for Apache Flink dan E-MapReduce terintegrasi secara mulus dengan Apache Paimon. Untuk informasi selengkapnya tentang Apache Paimon, lihat Apache Paimon.

Catatan penggunaan

MaxCompute hanya dapat membaca dari tabel eksternal Apache Paimon. Menulis ke tabel eksternal Paimon dan sinkronisasi perubahan skema otomatis tidak didukung.

Batasan tambahan:

Proyek dengan fitur schema yang diaktifkan tidak didukung.
Atribut clustering tidak didukung.
Kueri dan backtracking versi historis tidak didukung.

Prasyarat

Sebelum memulai, pastikan Anda telah memiliki:

Akun Alibaba Cloud dengan izin CreateTable pada MaxCompute. Untuk informasi selengkapnya, lihat Izin MaxCompute.
Proyek MaxCompute. Untuk informasi selengkapnya, lihat Buat proyek MaxCompute.
Object Storage Service (OSS) diaktifkan. Bucket OSS dan direktori file telah dibuat. Untuk informasi selengkapnya, lihat Buat bucket.
Realtime Compute for Apache Flink diaktifkan. Untuk informasi selengkapnya, lihat Aktifkan Realtime Compute for Apache Flink.

Deploy bucket OSS Anda di wilayah yang sama dengan proyek MaxCompute untuk menghindari masalah konektivitas lintas wilayah.

Langkah 1: Unggah plugin Apache Paimon

Unggah paimon_maxcompute_connector.jar ke proyek MaxCompute Anda menggunakan salah satu metode berikut.

Gunakan klien MaxCompute (odpscmd)

Hubungkan ke proyek MaxCompute Anda menggunakan odpscmd dan jalankan:

ADD JAR <path_to_paimon_maxcompute_connector.jar>;

Gunakan konsol DataWorks

Masuk ke konsol DataWorks. Di panel navigasi sebelah kiri, klik Workspace. Di halaman Workspaces, temukan workspace Anda dan pilih Shortcuts > Data Development di kolom Actions.
Di halaman DataStudio, klik Create dan pilih Create Resource > JAR.
Di kotak dialog Create Resource, konfigurasikan parameter, unggah paimon_maxcompute_connector.jar, lalu klik Create. Untuk informasi selengkapnya, lihat Langkah 1: Buat resource atau unggah resource yang sudah ada.
Setelah resource dibuat, klik ikon di bilah alat untuk commit resource ke lingkungan pengembangan.

Langkah 2: Buat tabel eksternal Apache Paimon menggunakan Flink

Pada langkah ini, Realtime Compute for Apache Flink menulis data Paimon ke OSS dan membuat katalog Paimon yang didukung oleh MaxCompute sebagai metastore. MaxCompute kemudian membaca tabel Paimon melalui pemetaan tabel eksternal.

Buat katalog Paimon

Masuk ke konsol Realtime Compute for Apache Flink dan buat skrip.

Di area pengeditan skrip pada tab Scripts, masukkan definisi katalog berikut, pilih kode tersebut, lalu klik Run.

Parameter koneksi MaxCompute

Parameter	Wajib	Deskripsi
`catalog name`	Ya	Nama katalog Paimon. Hanya boleh berisi huruf. Contoh: `catalogname`.
`type`	Ya	Jenis katalog. Atur ke `paimon`.
`metastore`	Ya	Jenis penyimpanan metadata. Atur ke `maxcompute`.
`warehouse`	Ya	Direktori gudang data di OSS, dalam format `oss://<bucket>/<object>`. Temukan nama bucket dan path objek di konsol OSS.
`maxcompute.endpoint`	Ya	Titik akhir MaxCompute untuk wilayah dan jenis jaringan Anda. Lihat Endpoints.
`maxcompute.project`	Ya	Nama proyek MaxCompute. Proyek dengan fitur schema yang diaktifkan tidak didukung.
`maxcompute.accessid`	Ya	ID AccessKey dari Akun Alibaba Cloud atau Pengguna RAM dengan izin MaxCompute. Dapatkan dari halaman AccessKey Pair.
`maxcompute.accesskey`	Ya	Rahasia AccessKey yang sesuai dengan ID AccessKey.

Parameter akses OSS

Parameter	Wajib	Deskripsi
`maxcompute.oss.endpoint`	Tidak	Titik akhir OSS untuk diakses oleh MaxCompute. Nilai default-nya adalah nilai `fs.oss.endpoint` jika tidak diatur. Bucket OSS dan proyek MaxCompute harus berada di wilayah yang sama. Kami merekomendasikan penggunaan titik akhir internal. Lihat Wilayah dan titik akhir.
`fs.oss.endpoint`	Tidak	Titik akhir OSS untuk diakses oleh Flink. Wajib diisi jika bucket OSS berada di wilayah berbeda dari workspace Flink atau dimiliki oleh Akun Alibaba Cloud yang berbeda. Konfigurasikan berdasarkan wilayah dan jenis jaringan bucket OSS Anda. Lihat Wilayah dan titik akhir.
`fs.oss.accessKeyId`	Tidak	ID AccessKey dengan izin baca dan tulis pada OSS. Wajib untuk akses cross-region atau cross-account. Dapatkan dari halaman AccessKey Pair.
`fs.oss.accessKeySecret`	Tidak	Rahasia AccessKey yang sesuai dengan `fs.oss.accessKeyId`. Wajib untuk akses cross-region atau cross-account.

CREATE CATALOG `<catalog name>` WITH (
  'type'                     = 'paimon',
  'metastore'                = 'maxcompute',
  'warehouse'                = '<warehouse>',
  'maxcompute.endpoint'      = '<maxcompute.endpoint>',
  'maxcompute.project'       = '<maxcompute.project>',
  'maxcompute.accessid'      = '<maxcompute.accessid>',
  'maxcompute.accesskey'     = '<maxcompute.accesskey>',
  'maxcompute.oss.endpoint'  = '<maxcompute.oss.endpoint>',
  'fs.oss.endpoint'          = '<fs.oss.endpoint>',
  'fs.oss.accessKeyId'       = '<fs.oss.accessKeyId>',
  'fs.oss.accessKeySecret'   = '<fs.oss.accessKeySecret>'
);

Buat dan isi tabel Paimon

Di area pengeditan skrip pada tab Scripts, jalankan pernyataan berikut untuk membuat tabel partisi bernama test_tbl. Tunggu pesan konfirmasi di tab Results.

CREATE TABLE `catalogname`.`default`.test_tbl (
  dt     STRING,
  id     BIGINT,
  data   STRING,
  PRIMARY KEY (dt, id) NOT ENFORCED
) PARTITIONED BY (dt);

Di tab Drafts halaman SQL Editor, buat draft SQL dengan pernyataan berikut dan deploy. Untuk informasi selengkapnya, lihat Kembangkan draft SQL.

Paimon melakukan commit data setelah setiap checkpoint selesai. Draft SQL memerlukan versi engine vvr-8.0.5-flink-1.17 atau lebih baru.

-- Atur checkpointing ke 10 detik untuk mempercepat commit data dalam contoh ini.
-- Di produksi, atur ke 1–10 menit berdasarkan kebutuhan latensi Anda.
SET 'execution.checkpointing.interval' = '10s';

INSERT INTO `catalogname`.`default`.test_tbl
VALUES ('2023-04-21', 1, 'AAA'),
       ('2023-04-21', 2, 'BBB'),
       ('2023-04-22', 1, 'CCC'),
       ('2023-04-22', 2, 'DDD');

Langkah 3: Kueri data dari MaxCompute

Jalankan pengaturan sesi berikut di klien MaxCompute (odpscmd) atau alat apa pun yang menjalankan SQL MaxCompute:
```
SET odps.sql.common.table.planner.ext.hive.bridge = true;
SET odps.sql.hive.compatible = true;
```

Kueri tabel eksternal Paimon:

SELECT * FROM test_tbl WHERE dt = '2023-04-21';

Output yang diharapkan:

+------------+------------+------------+
| id         | data       | dt         |
+------------+------------+------------+
| 1          | AAA        | 2023-04-21 |
| 2          | BBB        | 2023-04-21 |
+------------+------------+------------+