Realtime Compute for Apache Flink:AnalyticDB for PostgreSQL Erişimi

Sınırlamalar

• Realtime Compute for Apache Flink, sunucusuz moddaki AnalyticDB for PostgreSQL örneklerinden veri okuyamaz.

• AnalyticDB for PostgreSQL bağlayıcısı için Ververica Runtime (VVR) 6.0.0 veya sonraki bir sürüm gerekir.

• AnalyticDB for PostgreSQL V7.0 için VVR 8.0.1 veya sonraki bir sürüm gerekir.

Bunun yerine özel bir bağlayıcı kullanmak isterseniz bkz. Özel bağlayıcıları yönetme .

Önkoşullar

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

• Tam yönetimli bir Flink çalışma alanı. Bkz. Tam yönetimli Flink'i etkinleştirme.

• Bir AnalyticDB for PostgreSQL örneği ve ayrıcalıklı bir hesap. Bkz. Örnek oluşturma ve Ayrıcalıklı hesap oluşturma.

• AnalyticDB for PostgreSQL örneği ile tam yönetimli Flink çalışma alanının aynı sanal özel bulut (VPC) içinde bulunması.

Örnekler farklı VPC'lerde yer alıyorsa bkz. Tam yönetimli Flink, VPC'ler arasında bir hizmete nasıl erişir?

Adım 1: İzin listesini yapılandırma ve verileri hazırlama

1. AnalyticDB for PostgreSQL konsolu üzerinden oturum açın.

2. Tam yönetimli Flink çalışma alanınızın CIDR bloğunu AnalyticDB for PostgreSQL örneğinin izin listesine ekleyin.

   1. Tam yönetimli Flink çalışma alanınızın kullandığı vSwitch'in CIDR bloğunu bulun. Bkz. İzin listesi nasıl yapılandırılır?

   2. CIDR bloğunu AnalyticDB for PostgreSQL örneğinin izin listesine ekleyin. Bkz. Yordam.

   Örneğe internet üzerinden erişiyorsanız bunun yerine genel IP adresini ekleyin.

3. Örnek ayrıntıları sayfasında sağ üst köşede bulunan Log On to Database seçeneğine tıklayın ve kullanıcı adınız ile parolanızı girin. Ayrıntılar için bkz. İstemci araçlarını kullanarak bir örneğe bağlanma.

4. adbpg_dim_table adlı bir boyut tablosu oluşturun ve 50 satır örnek veri ekleyin.

   -- Create the dimension table
   CREATE TABLE adbpg_dim_table(
     id int,
     username text,
     PRIMARY KEY(id)
   );
   
   -- Insert 50 rows: id ranges from 1 to 50, username is "username" followed by the row number
   INSERT INTO adbpg_dim_table(id, username)
   SELECT i, 'username'||i::text
   FROM generate_series(1, 50) AS t(i);

   Eklenen verileri doğrulamak için SELECT * FROM adbpg_dim_table ORDER BY id; komutunu çalıştırın.

5. Flink'in çıktıyı yazacağı adbpg_sink_table adlı bir sonuç tablosu oluşturun.

   CREATE TABLE adbpg_sink_table(
     id int,
     username text,
     score int
   );

Adım 2: Akış taslağı oluşturma

1. Realtime Compute for Apache Flink konsolu üzerinden oturum açın, çalışma alanınızı bulun ve Actions sütunundaki Console seçeneğine tıklayın.

2. Sol gezinme bölmesinde Development > ETL yolunu izleyin. SQL Düzenleyici sayfasının sol üst köşesinde New seçeneğine tıklayın ve New Blank Stream Draft öğesini seçin.

3. New Draft iletişim kutusunda aşağıdaki parametreleri yapılandırın.

   Parametre	Açıklama	Örnek
   Name	Taslağın adı. Proje içinde benzersiz olmalıdır.	adbpg-test
   Location	Taslağın kaydedileceği klasör. Alt klasör oluşturmak için mevcut bir klasörün yanındaki simgeye tıklayın.	Draft
   Engine Version	Flink altyapı sürümü. Sürüm ayrıntıları ve yaşam döngüsü için bkz. Altyapı sürümleri.	vvr-8.0.1-flink-1.17

4. Create seçeneğine tıklayın.

Adım 3: Taslağı yazma ve dağıtma

1. Aşağıdaki SQL kodunu kod düzenleyiciye kopyalayın. Bu SQL kodu üç tablo tanımlar ve Datagen akışını AnalyticDB for PostgreSQL'deki kullanıcı verileriyle zenginleştiren bir arama birleşimini (lookup join) içerir.

   -- Source table: Datagen generates sequential IDs (1-50) and random scores (70-100).
   -- No changes needed in the WITH clause for this example.
   CREATE TEMPORARY TABLE datagen_source (
     id INT,
     score INT
   ) WITH (
     'connector' = 'datagen',
     'fields.id.kind' = 'sequence',
     'fields.id.start' = '1',
     'fields.id.end' = '50',
     'fields.score.kind' = 'random',
     'fields.score.min' = '70',
     'fields.score.max' = '100'
   );
   
   -- Dimension table: backed by AnalyticDB for PostgreSQL.
   -- Flink queries this table at processing time to look up usernames by ID.
   -- Replace the WITH clause values with your actual connection details.
   CREATE TEMPORARY TABLE dim_adbpg(
     id int,
     username varchar,
     PRIMARY KEY(id) NOT ENFORCED
   ) WITH (
     'connector' = 'adbpg',
     'url' = 'jdbc:postgresql://gp-2ze****3tysk255b5-master.gpdb.rds.aliyuncs.com:5432/flinktest',
     'tablename' = 'adbpg_dim_table',
     'username' = 'flinktest',
     'password' = '${secret_values.adb_password}',
     'maxRetryTimes' = '2',
     'cache' = 'lru',
     'cacheSize' = '100'
   );
   
   -- Result table: Flink writes enriched records here.
   -- Replace the WITH clause values with your actual connection details.
   CREATE TEMPORARY TABLE sink_adbpg (
     id int,
     username varchar,
     score int
   ) WITH (
     'connector' = 'adbpg',
     'url' = 'jdbc:postgresql://gp-2ze****3tysk255b5-master.gpdb.rds.aliyuncs.com:5432/flinktest',
     'tablename' = 'adbpg_sink_table',
     'username' = 'flinktest',
     'password' = '${secret_values.adb_password}',
     'maxRetryTimes' = '2',
     'conflictMode' = 'ignore',
     'retryWaitTime' = '200'
   );
   
   -- Lookup join: for each record from datagen_source, Flink looks up the matching
   -- row in dim_adbpg at the time the record is processed (PROCTIME()).
   INSERT INTO sink_adbpg
   SELECT ts.id, ts.username, ds.score
   FROM datagen_source AS ds
   JOIN dim_adbpg FOR SYSTEM_TIME AS OF PROCTIME() AS ts
   ON ds.id = ts.id;

   Arama birleşimi sözdizimi hakkında: FOR SYSTEM_TIME AS OF PROCTIME() ifadesi, Flink'e her kaynak kayıt işlendiği anda boyut tablosunu sorgulamasını söyler. Bu sayede her kayıt, işleme anında mevcut olan boyut verileriyle zenginleştirilir ve boyut tablosunda daha sonra yapılan değişiklikler önceden yazılmış sonuçları güncellemez.

2. Boyut tablosu ve sonuç tablosu için bağlantı parametrelerini güncelleyin. WITH yan tümcelerindeki yer tutucu değerleri kendi AnalyticDB for PostgreSQL bağlantı bilgilerinizle değiştirin. Datagen kaynak tablosunda değişiklik yapmanız gerekmez. Tam parametre referansı ve veri türü eşlemeleri için bkz. AnalyticDB for PostgreSQL bağlayıcısı.

   Parametre	Gerekli	Varsayılan	Açıklama
   url	Evet	—	jdbc:postgresql://<Internal endpoint>:<Port>/<Database name> biçimindeki JDBC URL'si. Bu bilgiyi AnalyticDB for PostgreSQL konsolundaki örneğin Database Connection sayfasında bulabilirsiniz.
   tablename	Evet	—	AnalyticDB for PostgreSQL veritabanındaki tablo adı.
   username	Evet	—	Veritabanına erişmek için kullanılan kullanıcı adı.
   password	Evet	—	Veritabanı hesabının parolası.
   targetSchema	Hayır	public	Şema adı. Tablonuz public şemasında değilse bu değeri belirtin.
   maxRetryTimes	Hayır	—	Yazma hatası sonrası yapılacak maksimum yeniden deneme sayısı.
   cache	Hayır	—	Boyut tablosu aramaları için önbellek ilkesi. Son erişilen girdilerin bellekte tutulması için lru olarak ayarlayın. LRU önbelleğe alma, veritabanı trafiğini azaltır ve arama aktarım hızını artırır ancak önbelleğe alınan girdiler eski olabilir. Bu, aktarım hızı ile veri güncelliği arasında bir dengedir; cacheSize parametresini ayarlayın ve etkinleştirmeden önce eski verilere toleransınızı gözden geçirin.
   cacheSize	Hayır	—	Önbelleğe alınacak maksimum girdi sayısı. Daha yüksek değerler veritabanı isteklerini azaltır ancak daha fazla bellek tüketir.
   conflictMode	Hayır	—	Bir yazma işleminin mevcut bir birincil anahtar veya dizinle çakışması durumunda alınacak eylem. Çakışan satırları atlamak için ignore olarak ayarlayın.
   retryWaitTime	Hayır	—	Yazma yeniden denemeleri arasında milisaniye cinsinden beklenecek süre.

3. SQL Düzenleyici sayfasının sağ üst köşesinde söz dizimini kontrol etmek için Validate seçeneğine tıklayın.

4. Deploy seçeneğine tıklayın.

5. O&M > Deployments sayfasında dağıtımınızı bulun ve Actions sütunundaki Start seçeneğine tıklayın.

Adım 4: Sonucu doğrulama

1. AnalyticDB for PostgreSQL konsolu üzerinden oturum açın.

2. Log On to Database seçeneğine tıklayın. Ayrıntılar için bkz. İstemciden bir örneğe bağlanma.

3. Flink'in sonuç tablosuna yazdığı kayıtları görüntülemek için aşağıdaki sorguyu çalıştırın.

   SELECT * FROM adbpg_sink_table ORDER BY id;

   Sonuç; her biri bir kullanıcı kimliği, boyut tablosundan gelen eşleşen kullanıcı adı ve 70 ile 100 arasında rastgele bir puan içeren 50 satır içermelidir.

   Sorgu, id, username ve score olmak üzere üç sütunlu 7 kayıt döndürür. 1'den 7'ye kadar olan satırlar sırasıyla username1'den username7'ye karşılık gelir ve puanlar 94, 79, 70, 93, 71, 82 ve 87 şeklindedir.

İlgili Kaynaklar

• Datagen bağlayıcısı

• AnalyticDB for PostgreSQL bağlayıcısı