Flink SQL Streaming node usage - DataWorks - Alibaba Cloud Dokümantasyon Merkezi

DataWorks Data Studio'daki Flink SQL Streaming node'ları, standart SQL kullanarak gerçek zamanlı işleme mantığını tanımlamanıza olanak tanır. Sağlam durum yönetimi, hata toleransı, olay zamanı ve işleme zamanı semantiğini destekler; Kafka ve HDFS gibi sistemlerle entegre çalışır. Bu konuda, gerçek zamanlı verileri işlemek üzere bir Flink SQL Streaming node'unun nasıl geliştirileceği, yapılandırılacağı ve çalıştırılacağı açıklanmaktadır.

Ön koşullar

Administration bölümünde Realtime Compute for Apache Flink için bir hesaplama kaynağı ilişkilendirilmiş olmalıdır. Daha fazla bilgi için bkz. Hesaplama motorunu bağlama.
Bir Flink SQL Streaming node'u oluşturulmuş olmalıdır. Daha fazla bilgi için bkz. Zamanlama iş akışı için node oluşturma.
DataWorks'ün Realtime Compute for Apache Flink API'lerini çağırırken kullandığı RAM kullanıcısına veya RAM rolüne gerekli OpenAPI izinleri verilmiş olmalıdır. Bu izinler, DataWorks'ün node görevlerini bir Flink kümesine göndermesine ve dağıtmasına olanak tanır.
```
{
  "Version": "1",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["stream:CreateDeployment", "stream:UpdateDeployment", "stream:GetDeployment", "stream:DeleteDeployment"],
      "Resource": ["*"]
    }
  ]
}
```

Sınırlamalar

Bu node bir iş akışında kullanılamaz; bağımsız bir node olarak geliştirilmeli ve çalıştırılmalıdır.
Yalnızca sunucusuz kaynak grupları desteklenir. Zamanlama için kullanılan eski özel kaynak grupları desteklenmez.

Adım 1: Flink SQL Streaming node'unu geliştirme

Flink SQL Streaming node'u düzenleme sayfasında node görevini geliştirin.

SQL kodunu geliştirme

SQL düzenleyicisinde, ${değişken_adı} biçimini kullanarak değişkenler tanımlayabilirsiniz. Bu değişkenlere, zamanlama senaryolarında parametreleri dinamik olarak iletmek için Real-Time configuration panelindeki Script Parameters bölümünde değerler atayın. Örneğin:

--Create the source table datagen_source. CREATE TEMPORARY TABLE datagen_source( name VARCHAR ) WITH ( 'connector' = 'datagen' ); --Create the result table blackhole_sink. CREATE TEMPORARY TABLE blackhole_sink( name VARCHAR ) WITH ( 'connector' = 'blackhole' ); --Insert data from the source table into the result table. INSERT INTO blackhole_sink SELECT name FROM datagen_source WHERE LENGTH(name) > ${name_length};

Not

Bu örnekte, name_length parametresinin değeri 5 olarak ayarlanmıştır. Bu parametre, yalnızca ad uzunluğu 5 karakterden büyük olan kayıtları işlemek için verileri filtreler.

Adım 2: Flink SQL Streaming node'unu yapılandırma

İş gereksinimlerinize göre Flink SQL Streaming node'u için aşağıdaki parametreleri yapılandırın.

Flink kaynaklarını yapılandırma

Real-Time configuration panelinin Flink resource information bölümünde, seçilen Resource Mode değerine göre aşağıdaki parametreleri yapılandırın. Daha fazla bilgi için bkz. Flink kaynaklarını yapılandırma.

Parametre	Açıklama
Flink cluster	Administration bölümünde ilişkilendirilen tam yönetimli Flink hesaplama kaynağı.
Flink engine version	Kullanılacak motor sürümü. İhtiyaçlarınıza göre bir sürüm seçin.
Resource Group	Flink ile ağ bağlantısı olan bir sunucusuz kaynak grubu seçin.
Resource Mode aşağıdaki iki modu destekler. Daha fazla bilgi için bkz. Flink kaynaklarını yapılandırma. Basic mode (varsayılan): Yeni başlayanlar ve basit senaryolar için uygundur. Flink işlerini hızlıca başlatmak için varsayılan ve basitleştirilmiş ayarları kullanır. Expert mode: Deneyimli kullanıcılar için gelişmiş seçenekler sunar; karmaşık veya yüksek performans gerektiren senaryolarda performans ve kaynakların hassas şekilde ayarlanmasını sağlar. Seçtiğiniz kaynak moduna göre parametreleri yapılandırın. Flink mimarisini anlamak, parametreleri daha etkili bir şekilde yapılandırmanıza yardımcı olur. Daha fazla bilgi için bkz. Flink Architecture \| Apache Flink.
Basic mode
Job Manager CPU	JobManager'ın kararlı çalışması için en az 0,5 CPU çekirdeği ve 2 GiB bellek gerekir. Önerilen yapılandırma 1 CPU çekirdeği ve 4 GiB bellektir; maksimum 16 CPU çekirdeğine kadar çıkılabilir. Küme ölçeğine ve iş karmaşıklığına göre ayarlayın.
Job Manager Memory	JobManager belleği, zamanlama ve yönetim kapasitesini etkiler. Önerilen aralık 2 GiB ile 64 GiB arasındadır. Küme ölçeğine ve iş gereksinimlerine göre ayarlayın.
Task Manager CPU	TaskManager CPU'su, görev işleme kapasitesini etkiler. En az 0,5 CPU çekirdeği ve 2 GiB bellek önerilir; tercih edilen yapılandırma 1 CPU çekirdeği ve 4 GiB bellektir. Maksimum 16 CPU çekirdeğine kadar çıkılabilir. Gereksinimlerinize göre ayarlayın.
Task Manager Memory	TaskManager belleği, veri hacmini ve işleme performansını belirler. Bellek boyutu en az 2 GiB olmalı ve 64 GiB'a kadar ayarlanabilir.
Concurrency	Bir Flink işinde paralel çalışan görev yürütme sayısı. Daha yüksek eşzamanlılık, işleme hızını ve kaynak kullanımını artırır. Bu değeri küme kaynaklarına ve iş özelliklerine göre ayarlayın.
Number of slots per TaskManager	Her TaskManager'daki yuva sayısı; bu değer, kaç görevin paralel çalışabileceğini belirler. Kaynak kullanımını ve paralel işlemeyi optimize etmek için yuvaları ayarlayın.
Expert mode
Job Manager CPU	JobManager'ın kararlı çalışması için en az 0,25 CPU çekirdeği ve 1 GiB bellek gerekir; maksimum 16 CPU çekirdeğine kadar çıkılabilir. Küme ölçeğine ve iş karmaşıklığına göre ayarlayın.
Job Manager Memory	JobManager belleği, zamanlama ve yönetim kapasitesini etkiler. Önerilen aralık 1 GiB ile 64 GiB arasındadır. Küme ölçeğine ve iş gereksinimlerine göre ayarlayın.
Number of slots per TaskManager	Her TaskManager'daki yuva sayısı; bu değer, kaç görevin paralel çalışabileceğini belirler. Kaynak kullanımını ve paralel işlemeyi optimize etmek için yuvaları ayarlayın.
Multiple SSG mode	Varsayılan olarak tüm operatörler tek bir yuva paylaşım grubunu paylaşır, bu nedenle her operatör için kaynakları ayrı ayrı yapılandıramazsınız. Her operatöre kendi bağımsız yuvasını atamak ve ardından ilgili yuvada kaynakları yapılandırmak için Multiple SSG mode özelliğini etkinleştirin.

(İsteğe bağlı) Komut dosyası parametrelerini yapılandırma

Sağ gezinme bölmesindeki Real-Time configuration panelinin Script Parameters bölümünde, kodunuzda dinamik olarak kullanmak üzere Add parameters seçeneğine tıklayın ve Parameter name ile Parameter Value alanlarını düzenleyin.

(İsteğe bağlı) Flink çalışma parametrelerini yapılandırma

Sağ gezinme bölmesindeki Real-Time configuration panelinin Flink running parameters bölümünde aşağıdaki parametreleri yapılandırın. Daha fazla bilgi için bkz. Flink çalışma parametrelerini yapılandırma.

Parametre	Açıklama
System Checkpoint Interval	Flink'in periyodik kontrol noktaları gerçekleştirdiği zaman aralığı. Daha kısa bir aralık, hata kurtarma süresini azaltır ancak sistem yükünü artırır. Boş bırakılırsa kontrol noktaları devre dışı bırakılır.
Minimum time interval between two system checkpoints	Ardışık kontrol noktaları arasındaki minimum bekleme süresi; sık yapılan kontrol noktalarının performansı etkilemesini önler. Bu, kontrol noktaları için maksimum paralellik 1 olduğunda iki kontrol noktası arasında minimum bir boşluk olmasını sağlar.
State Data Expiration Time	Durum verilerinin erişilmeden veya güncellenmeden saklanabileceği maksimum süre. Varsayılan 36 saattir; bu sürenin sonunda durum bilgileri otomatik olarak sona erer ve depolama ile kaynak kullanımını optimize etmek için temizlenir. Önemli Bu varsayılan değer cloud en iyi uygulamalarına dayanır ve açık kaynaklı varsayılandan (0, yani durumun asla sona ermemesi) farklıdır.
Others	Ek Flink çalışma parametreleri. Örneğin: `taskmanager.network.memory.max:4g`. Not Parametre yapılandırmaları hakkında daha fazla bilgi için bkz. Flink çalışma parametrelerini yapılandırma.

Görev yapılandırıldıktan sonra, node görevini kaydetmek için Save seçeneğine tıklayın.

Adım 3: (İsteğe bağlı) Flink SQL Streaming node'unu hata ayıklama

Node'u üretime dağıtmadan önce, yüklenen sahte verilere karşı node kodunu çalıştırarak hata ayıklaması yapabilirsiniz. Bu sayede, görevi Operation Center'a dağıtmadan SQL mantığını ve yukarı/aşağı akış verilerini doğrulayabilirsiniz.

Not

Hata ayıklama özelliği bir izin listesi üzerinden sunulmaktadır. Bu özelliği kullanmak için erişim talebinde bulunmak üzere bir destek bileti oluşturun.

Flink kaynak bilgilerini yapılandırma

Node düzenleme sayfasının sağ tarafındaki Run Configuration panelinin Flink resource information bölümünde, aşağıdaki tabloda açıklandığı gibi parametreleri yapılandırın.

Parametre	Açıklama
Flink Debug Cluster	Hata ayıklama görevini çalıştırmak için kullanılan Flink oturum kümesi. Zorunludur. Açılan listede, mevcut hesaplama kaynağı altındaki mevcut oturum kümeleri ve çalışma durumları gösterilir. Yalnızca Running durumunda olan kümeler seçilebilir. Listede kullanılabilir küme yoksa, bir oturum kümesi oluşturmak üzere Realtime Compute for Apache Flink konsoluna gitmek için Create Cluster seçeneğine tıklayın.
Flink Engine Version	Seçilen oturum kümesinin Flink motor sürümü. Küme seçimine göre otomatik olarak görüntülenir.
Timeout	Tek bir hata ayıklama görevi için maksimum çalışma süresi (dakika cinsinden). Varsayılan: 30 dakika. Hata ayıklama görevi bu sürenin sonunda otomatik olarak durur.

Not

Mevcut node için hesaplama kaynağını değiştirdikten sonra, seçilen Flink Debug Cluster ve yüklenen hata ayıklama verileri temizlenir. Bir kümeyi yeniden seçmeniz ve verileri yeniden yüklemeniz gerekir.

Hata ayıklama verilerini hazırlama

Run Configuration panelinin Debug Data bölümünde, kodunuzda başvurulan kaynak tablolar için sahte verileri hazırlayın.

Generate Template seçeneğine tıklayın. Sistem, mevcut SQL'de başvurulan kaynak tabloları ayrıştırır ve aşağıdaki listede karşılık gelen tablo adı kayıtları oluşturur. Daha önce yüklenen veriler temizlenmez.
Bir tablo adı kaydının Actions sütununda, kaynak tablonun şemasına uygun bir CSV şablonu indirmek için Download Template seçeneğine tıklayın.
Şablondaki sütun sırasına göre hata ayıklama verilerini yerel olarak doldurun ve dosyayı CSV formatında kaydedin.
Bir tablo adı kaydının Actions sütununda, tamamlanan CSV dosyasını seçip yüklemek için Upload seçeneğine tıklayın. Yükleme başarılı olduktan sonra, Status sütununda Enabled durumu görüntülenir.
(İsteğe bağlı) Yükleme başarılı olduktan sonra, alt paneldeki veri içeriğini görüntülemek için Preview seçeneğine tıklayabilirsiniz. Verileri değiştirmek için mevcut verilerin üzerine yazmak amacıyla yeni bir CSV dosyası yükleyin.
Belirli bir kaynak tablonun sahte verilerinin mevcut hata ayıklama oturumuna katılmasını istemiyorsanız, durumu Disabled olarak değiştirmek için Disable seçeneğine tıklayın. Yeniden etkinleştirmek için Enable seçeneğine tıklayın. Hata ayıklama oturumunda yalnızca Enabled durumundaki veriler kullanılır.

Not

Hata ayıklama verilerini yüklemeden önce öncelikle bir Flink Debug Cluster seçin. Aksi takdirde, önce bir hesaplama kaynağı seçmeniz gerektiğine dair bir uyarı alırsınız.

Hata ayıklama verileri yalnızca CSV formatını destekler ve maksimum dosya boyutu 1 MB'tır. CSV dosyasının ilk satırı sütun adlarını içermelidir ve UTF-8 kodlaması önerilir.

Hata ayıklama görevini çalıştırma

Hata ayıklama verileri hazırlandıktan sonra, düzenleyici araç çubuğundaki Run düğmesine tıklayın (veya F8 tuşuna basın). Sistem, kodu, sahte verileri ve Flink kaynak bilgilerini seçilen oturum kümesinde yürütülmek üzere gönderir.

Kodunuz ${değişken_adı} biçiminde parametreler kullanıyorsa, değişkenlere Script Parameters bölümünde değer atadığınızdan emin olun. Hata ayıklama sırasında sistem, gönderimden önce koddaki yer tutucuları atanan değerlerle değiştirir.

Hata ayıklama sonuçlarını görüntüleme

Hata ayıklama görevi çalıştıktan sonra, node'un altındaki sonuç alanında aşağıdaki bilgiler gösterilir:

Code: Flink motoruna gönderilen SQL kodu (değişken değiştirme uygulanmış halde).
Logs: Çalışma zamanı günlükleri ve hata bilgileri.
Query results: Hata ayıklama görevinden çıkan veriler.

Adım 4: Flink SQL Streaming node'unu başlatma

Flink SQL Streaming node'unu dağıtın.

Görevin çalışabilmesi için Operation Center'a dağıtılması gerekir. Node'u dağıtmak için ekrandaki talimatları izleyin. Daha fazla bilgi için bkz. Node dağıtma.

Not
Bu işlem aynı zamanda görevi Flink VVP çalışma alanına da dağıtır. DataWorks üzerinden dağıtılan görevleri Flink VVP Operation Center > Job O&M bölümünde görüntüleyebilirsiniz.
Flink SQL Streaming node'unu başlatın.

Görev dağıtıldıktan sonra, Deploy to production environment altındaki Go to operation and maintenance seçeneğine tıklayın. Operation Center'da Node O&M > Real-time Task O&M > Real-time computing tasks yolunu izleyin, başlatmak istediğiniz görevi bulun ve gerçek zamanlı görevi başlatıp çalışma durumunu görüntülemek için Operation sütunundaki Start seçeneğine tıklayın.