Realtime Compute for Apache Flink:Iceberg kataloglarını yönetin

Ön koşullar

Başlamadan önce aşağıdakilere sahip olduğunuzdan emin olun:

• Ververica Runtime (VVR) 11,1 veya sonraki bir sürümü çalıştıran bir Flink çalışma alanı

• Flink çalışma alanınızla aynı bölgede bulunan bir DLF kataloğu. Bkz. DLF ile hızlı başlangıç

Sınırlamalar

Iceberg DLF kataloğu oluşturun

Flink'te bir Iceberg kataloğu oluşturmak, mevcut DLF kataloğunuzla bir eşleme kaydeder. Bu işlem DLF'deki verileri oluşturmaz veya değiştirmez. Iceberg REST aracılığıyla DLF kataloğunda oluşturulan tüm tablolar Iceberg tablolarıdır.

1. Realtime Compute for Apache Flink konsoluna giriş yapın.

2. Çalışma alanınızın Actions sütununda Console seçeneğine tıklayın.

3. Sol gezinme menüsünde Development > Scripts yolunu izleyin ve yeni bir betik oluşturun.

4. SQL düzenleyicisine aşağıdaki ifadeyi yapıştırın. Sağ alt köşede Environment seçeneğine tıklayın ve VVR 11,2,0 veya sonraki bir sürümü çalıştıran bir oturum kümesi seçin. Ardından ifadeyi çalıştırın.

   CREATE CATALOG `catalog_name`
   WITH (
       'type'              = 'iceberg',
       'catalog-type'      = 'rest',
       'token.provider'    = 'dlf',
       'uri'               = 'http://cn-hangzhou-vpc.dlf.aliyuncs.com/iceberg',
       'warehouse'         = 'iceberg_test',
       'rest.signing-region' = 'cn-hangzhou',
       'io-impl'           = 'org.apache.iceberg.rest.DlfFileIO'
   );

   uri, warehouse ve rest.signing-region değerlerini kendi değerlerinizle değiştirin. Bölgeye göre kullanılabilir URI listesini görmek için bkz. Iceberg REST uç noktası. Aşağıdaki tabloda her bir seçenek açıklanmaktadır.

   Seçenek	Gerekli mi?	Açıklama	Örnek
   type	Evet	iceberg olarak ayarlayın.	iceberg
   catalog-type	Evet	rest olarak ayarlayın.	rest
   token.provider	Evet	dlf olarak ayarlayın. DLF tabanlı kimlik doğrulamayı yapılandırır.	dlf
   uri	Evet	DLF kataloğunuz için Iceberg REST uç noktası. URI bölgeye göre değişir. Bkz. Iceberg REST uç noktası.	http://ap-southeast-1-vpc.dlf.aliyuncs.com/iceberg
   warehouse	Evet	DLF kataloğunuzun adı.	iceberg_test
   rest.signing-region	Evet	DLF'nin dağıtıldığı bölge kimliği.	cn-hangzhou
   io-impl	Evet	org.apache.iceberg.rest.DlfFileIO olarak ayarlayın.	org.apache.iceberg.rest.DlfFileIO

Katalog görüntüleyin veya silin

1. Realtime Compute for Apache Flink konsolunda, çalışma alanınızın Actions sütununda Console seçeneğine tıklayın.

2. Data Management sayfasına gidin ve Catalog List bölümünü açın.

   • Bir katalogdaki veritabanlarını ve tabloları görüntülemek için katalog adının yanındaki View seçeneğine tıklayın.

   • Bir kataloğu silmek için Actions sütunundaki Delete seçeneğine tıklayın.

     Önemli

     Katalog silindiğinde yalnızca Flink projenizdeki Data Management bölümünden kaydı kaldırılır. DLF'deki temel Iceberg veri dosyaları etkilenmez. Tabloları yeniden kullanmak için kataloğu tekrar kaydedin.

   Alternatif olarak, Veri Sorgulama sayfasında aşağıdaki ifadeyi çalıştırın:

   DROP CATALOG <catalog_name>;

Iceberg veritabanlarını yönetin

Aşağıdaki ifadeleri Veri Sorgulama sayfasında çalıştırın: kodu seçin ve Run seçeneğine tıklayın.

-- Replace my-catalog with your catalog name.
USE CATALOG `my-catalog`;

-- Replace my_db with the database name you want to create.
CREATE DATABASE `my_db`;

-- Replace my-catalog with your catalog name.
USE CATALOG `my-catalog`;

-- Delete a database only if it is empty.
DROP DATABASE `my_db`;

-- Delete a database and all tables inside it.
DROP DATABASE `my_db` CASCADE;

Iceberg tablolarını yönetin

Bir Iceberg kataloğu kaydedildikten sonra, Veri Tanımlama Dili (DDL) bildirmeden bir işte bu kataloğun tablolarına başvurabilirsiniz. Tam üç parçalı adı <catalog>.<database>.<table> şeklinde kullanın veya önce USE CATALOG ve USE <database> ifadelerini çalıştırıp ardından tabloya yalnızca adıyla başvurun.

-- Replace my-catalog with your Iceberg catalog name.
-- Replace my_db with your database name.
-- Replace my_tbl with your table name.
CREATE TABLE `my-catalog`.`my_db`.`my_tbl` (
  dt           STRING,
  shop_id      BIGINT,
  user_id      BIGINT,
  num_orders   INT,
  total_amount INT,
  PRIMARY KEY (dt, shop_id, user_id) NOT ENFORCED
) PARTITIONED BY (dt);

ALTER TABLE my_table RENAME TO my_table_new;

-- Add c1 (INT) and c2 (STRING) to the end of the table.
ALTER TABLE my_table ADD (c1 INT, c2 STRING);

-- Rename column c0 to c1.
ALTER TABLE my_table RENAME c0 TO c1;

-- Delete columns c1 and c2.
ALTER TABLE my_table DROP (c1, c2);

ALTER TABLE my_table MODIFY buy_count BIGINT COMMENT 'this is buy count';

-- Move col_a to the beginning of the table.
ALTER TABLE my_table MODIFY col_a DOUBLE FIRST;

-- Move col_a immediately after col_b.
ALTER TABLE my_table MODIFY col_a DOUBLE AFTER col_b;

-- Temporarily enable upsert mode when writing to a table.
INSERT INTO tableName /*+ OPTIONS('upsert-enabled'='true') */
SELECT ...;

-- Temporarily set the source read interval to 10 seconds.
SELECT * FROM my_table /*+ OPTIONS('monitor-interval'='10s') */;

-- Replace my-catalog, my_db, and my_tbl with your values.
DROP TABLE `my-catalog`.`my_db`.`my_tbl`;