Realtime Compute for Apache Flink:Paimon hızlı başlangıç: Temel özellikler

Önkoşullar

• Bir RAM kullanıcısı veya RAM rolü kullanıyorsanız, Realtime Compute for Apache Flink konsolu erişimi için gerekli izinlere sahip olduğundan emin olun. Daha fazla bilgi için bkz. İzin yönetimi.

• Bir Flink çalışma alanı oluşturdunuz. Daha fazla bilgi için bkz. Realtime Compute for Apache Flink'i etkinleştirme.

• Object Storage Service (OSS) hizmetini etkinleştirdiniz ve Standart depolama sınıfında bir paket oluşturdunuz. Daha fazla bilgi için bkz. OSS konsolunu kullanarak ilk adımları atma. OSS, Paimon tablolarının veri dosyaları ve meta veri dosyaları dahil olmak üzere dosyalarını depolamak için kullanılır.

• Paimon tabloları yalnızca Ververica Runtime (VVR) 8.0.5 veya sonraki sürümlerde desteklenir.

1. Adım: Paimon kataloğu oluşturma

1. Betik düzenleyicisine gidin.

   1. Realtime Compute for Apache Flink konsolu üzerinde oturum açın.

   2. Hedef çalışma alanının Actions sütununda, Console seçeneğini tıklayın.

   3. Sol gezinme bölmesinde, Data Development > Data Query seçeneklerini tıklayın. Query Script sekmesinde yeni bir Query Script oluşturun.

2. Düzenleyicide bir Paimon kataloğu oluşturmak için aşağıdaki kodu girin.

   -- my-catalog is a custom catalog name.
   CREATE Catalog `my-catalog` WITH (
     'type' = 'paimon',
     'metastore' = 'filesystem',
     'warehouse' = '<warehouse>',
     'fs.oss.endpoint' = '<fs.oss.endpoint>',
     'fs.oss.accessKeyId' = '<fs.oss.accessKeyId>',
     'fs.oss.accessKeySecret' = '<fs.oss.accessKeySecret>'
   );

   Aşağıdaki tabloda parametreler açıklanmaktadır.

   Parameter	Description	Required	Remarks
   type	Katalog türü.	Evet	Değer 'paimon' olmalıdır.
   metastore	Meta veri deposu türü.	Evet	Bu örnekte filesystem kullanılır. Diğer türler hakkında daha fazla bilgi için bkz. Paimon kataloglarını yönetme.
   warehouse	OSS paketinizdeki depo dizini.	Evet	Biçim oss://<bucket>/<object> şeklindedir; burada: bucket: Oluşturduğunuz OSS paketinin adı. object: Verileri depolamak istediğiniz yol. Paket ve nesne adlarınızı OSS konsolu üzerinde bulabilirsiniz.
   fs.oss.endpoint	OSS hizmetinin uç noktası.	Hayır	Warehouse tarafından belirtilen OSS paketi, Flink çalışma alanınızdan farklı bir bölgedeyse veya başka bir Alibaba Cloud hesabına aitse gereklidir. Daha fazla bilgi için bkz. Bölgeler ve Uç Noktalar. Not Bir Paimon tablosunu OSS-HDFS'de depolamak istiyorsanız fs.oss.endpoint, fs.oss.accessKeyId ve fs.oss.accessKeySecret parametrelerini belirtmeniz gerekir. fs.oss.endpoint değeri cn-<region>.oss-dls.aliyuncs.com biçimindedir; örneğin cn-hangzhou.oss-dls.aliyuncs.com.
   fs.oss.accessKeyId	OSS'e okuma ve yazma izni olan bir Alibaba Cloud hesabının veya RAM kullanıcısının AccessKey ID'si.	Hayır	Warehouse tarafından belirtilen OSS paketi, Flink çalışma alanınızdan farklı bir bölgedeyse veya başka bir Alibaba Cloud hesabına aitse gereklidir. AccessKey çiftini nasıl alacağınız hakkında bilgi için bkz. AccessKey çifti oluşturma.
   fs.oss.accessKeySecret	OSS'e okuma ve yazma izni olan Alibaba Cloud hesabının veya RAM kullanıcısının AccessKey gizli anahtarı.	Hayır

3. Paimon kataloğunu oluşturan kodu seçin ve soldaki Run seçeneğini tıklayın.

   The following statement has been executed successfully! iletisi, kataloğun oluşturulduğunu gösterir.

2. Adım: Paimon veritabanı ve tablo oluşturma

1. Scripts sayfasında my_db adlı bir Paimon veritabanı ve my_tbl adlı bir Paimon tablosu oluşturmak için aşağıdaki kodu girin.

   CREATE DATABASE `my-catalog`.`my_db`;
   CREATE TABLE `my-catalog`.`my_db`.`my_tbl` (
     dt STRING,
     id BIGINT,
     content STRING,
     PRIMARY KEY (dt, id) NOT ENFORCED
   ) PARTITIONED BY (dt) WITH (
     'changelog-producer' = 'lookup'  
   );

   Not

   Paimon tablosunun akış tüketimini daha sonra etkinleştirmek için bu örnekte WITH yan tümcesinde changelog-producer değeri lookup olarak ayarlanır; bu, değişiklik günlüğünü oluşturmak için arama stratejisini kullanır. Daha fazla bilgi için bkz. Değişiklik günlüğü üretimi.

2. Paimon veritabanını ve Paimon tablosunu oluşturan kodu seçin, ardından soldaki Run seçeneğini tıklayın.

   The following statement has been executed successfully! iletisi döndürülürse my_db veritabanı ve my_tbl tablosu oluşturulmuş demektir.

3. Adım: Paimon tablosuna veri yazma

1. Development > ETL sayfasının Drafts sekmesinde bir blank stream draft oluşturun. Daha fazla bilgi için bkz. SQL taslağı geliştirme. Aşağıdaki INSERT ifadesini SQL düzenleyicisine kopyalayın.

   -- Paimon commits data only after each checkpoint is complete.
   -- The checkpoint interval is shortened to 10s to commit data faster.
   -- In a production environment, typically set the checkpoint interval and the minimum interval between checkpoints from 1 to 10 minutes, based on your latency requirements.
   SET 'execution.checkpointing.interval'='10s';
   INSERT INTO `my-catalog`.`my_db`.`my_tbl` VALUES ('20240108',1,'apple'), ('20240108',2,'banana'), ('20240109',1,'cat'), ('20240109',2,'dog');

2. SQL düzenleyicisinin sağ üst köşesinde Deploy seçeneğini tıklayın. Deploy draft iletişim kutusunda parametreleri istendiği gibi yapılandırın ve Confirm seçeneğini tıklayın.

3. O&M > Deployments sayfasında hedef dağıtımı bulun ve Actions sütunundaki Start seçeneğini tıklayın. Initial Mode seçeneğini belirleyin ve Start seçeneğini tıklayın.

   İş durumu FINISHED olarak değiştiğinde veriler yazılmış demektir.

4. Adım: Paimon tablosundan veri akışı tüketme

1. Bir blank stream draft oluşturun. print bağlayıcısını kullanarak my_tbl tablosundaki tüm verileri günlüklerde çıktılanmak üzere aşağıdaki SQL kodunu SQL düzenleyicisine kopyalayın.

   CREATE TEMPORARY TABLE Print (
     dt STRING,
     id BIGINT,
     content STRING
   ) WITH (
     'connector' = 'print'
   );
   INSERT INTO Print SELECT * FROM `my-catalog`.`my_db`.`my_tbl`;

2. SQL düzenleyicisinin sağ üst köşesinde Deploy seçeneğini tıklayın. Deploy draft iletişim kutusunda parametreleri istendiği gibi yapılandırın ve Confirm seçeneğini tıklayın.

3. O&M > Deployments sayfasında, hedef dağıtımın Actions sütunundaki Start seçeneğini tıklayın. Initial Mode seçeneğini belirleyin ve Start seçeneğini tıklayın.

4. Dağıtım ayrıntıları sayfasında Flink hesaplama sonuçlarını görüntüleyin.

   1. O&M > Deployments sayfasında hedef dağıtımın adını tıklayın.

   2. Logs sekmesinde, Running Logs altında Running Task Managers sekmesine gidin ve Path, ID sütunundaki bir görev bağlantısını tıklayın.

   3. Tüketilen Paimon verilerini görüntülemek için Stdout seçeneğini tıklayın.

   Stdout çıktısı, Paimon tablosu için INSERT kayıtlarını gösterir: +I[20240108, 1, apple], +I[20240108, 2, banana], +I[20240109, 1, cat] ve +I[20240109, 2, dog]. Bu durum, akış tüketiminin başarılı olduğunu doğrular.

5. Adım: Paimon tablosundaki verileri güncelleme

1. Bir blank stream draft oluşturun ve aşağıdaki SQL kodunu SQL düzenleyicisine kopyalayın.

   SET 'execution.checkpointing.interval' = '10s';
   INSERT INTO `my-catalog`.`my_db`.`my_tbl` VALUES ('20240108', 1, 'hello'), ('20240109', 2, 'world');

2. SQL düzenleyicisinin sağ üst köşesinde Deploy seçeneğini tıklayın. Deploy draft iletişim kutusunda parametreleri istendiği gibi yapılandırın ve Confirm seçeneğini tıklayın.

3. O&M > Deployments sayfasında, hedef dağıtımın Actions sütunundaki Start seçeneğini tıklayın. Initial Mode seçeneğini belirleyin ve Start seçeneğini tıklayın.

   İş durumu FINISHED olarak değiştiğinde veriler güncellenmiş demektir.

4. Adımdaki işin Deployments sayfasının Stdout sekmesinde Paimon tablosuna güncellenen verileri görüntüleyin.

   Running Task Managers sekmesinde hedef TaskManager'ı tıklayın ve Stdout alt sekmesini açın. Paimon tablosu için CDC kayıtları görünür: -U[20240108, 1, apple], +U[20240108, 1, hello], -U[20240109, 2, dog] ve +U[20240109, 2, world]. -U güncellemeden önceki kaydı, +U ise güncellemeden sonraki kaydı gösterir.

6. Adım (İsteğe bağlı): İşi iptal etme ve kaynakları temizleme

Test tamamlandıktan sonra akış tüketimi işini iptal edin ve kaynakları temizleyin.

1. O&M > Deployments sayfasında, hedef dağıtımın Actions sütunundaki Cancel seçeneğini tıklayarak işi durdurun.

2. Scripts sekmesindeki düzenleyicide Paimon veri dosyalarını ve Paimon kataloğunu silmek için aşağıdaki kodu girin.

   DROP DATABASE `my-catalog`.`my_db` CASCADE; -- Deletes all data files for the Paimon database from OSS.
   DROP CATALOG `my-catalog`; -- Removes the Paimon catalog from the Realtime Compute for Apache Flink console metadata but does not affect the data files in OSS.

   The following statement has been executed successfully! iletisi, Paimon veri dosyalarının ve kataloğun silindiğini gösterir.

İlgili konular

• Paimon tablosuna veri yazma veya tablodan veri tüketme hakkında daha fazla bilgi için bkz. Paimon tablolarına veri yazma ve tablolardan veri tüketme.

• Paimon tablo şemasını değiştirmek (sütun ekleme veya sütun türlerini değiştirme gibi) ya da tablo parametrelerini geçici olarak değiştirmek için bkz. Tablo şemasını değiştirme.

• Paimon birincil anahtar tabloları ve Append Scalable tabloları için yaygın optimizasyonlar hakkında bilgi için bkz. Paimon performans optimizasyonu.

• Paimon ile ilgili sık sorulan sorular için bkz. Bağlayıcılar.