ApsaraMQ for Kafka:AnalyticDB sink connector oluşturma

Ön koşullar

Başlamadan önce aşağıdaki ön koşulların karşılandığından emin olun:

• Message Queue for Apache Kafka

  • Message Queue for Apache Kafka örneği için connector özelliğini etkinleştirin. Daha fazla bilgi için bkz. Connector'ları etkinleştirme.

  • Message Queue for Apache Kafka örneği için bir kaynak konusu oluşturun. Daha fazla bilgi için bkz. Adım 1: Konu oluşturma.

• Function Compute

  • Function Compute'u etkinleştirme.

• AnalyticDB for MySQL ve AnalyticDB for PostgreSQL

  • AnalyticDB for MySQL: AnalyticDB for MySQL konsolunda bir küme oluşturun, bir veritabanı hesabı oluşturun, kümeye bağlanın ve ardından bir veritabanı oluşturun. Daha fazla bilgi için bkz. Küme oluşturma, Veritabanı hesabı oluşturma, Kümeye bağlanma ve Veritabanı oluşturma.

  • AnalyticDB for PostgreSQL: AnalyticDB for PostgreSQL konsolunda bir örnek oluşturun, bir veritabanı hesabı oluşturun ve ardından veritabanında oturum açın. Daha fazla bilgi için bkz. Örnek oluşturma, Hesapları oluşturma ve yönetme ve Örneğe bağlanma.

Kullanım notları

• Tüm hizmetler aynı bölgede bulunduğunda, yalnızca AnalyticDB for MySQL aracılığıyla Message Queue for Apache Kafka örneğinin bir kaynak konusundan AnalyticDB for MySQL veya AnalyticDB for PostgreSQL'e veri aktarabilirsiniz. Connector sınırlamaları hakkında daha fazla bilgi için bkz. Sınırlamalar.

• Bu özellik Function Compute tarafından desteklenmektedir. Function Compute ücretsiz bir kota sağlar. Ücretsiz kotayı aşan kaynak kullanımı için ücretlendirme uygulanır. Daha fazla bilgi için Function Compute'un Faturalandırma genel bakışına bakın.

• Sorunları gidermek amacıyla Function Compute'da işlev çağrı günlüklerini sorgulayabilirsiniz. Daha fazla bilgi için bkz. Günlük kaydı yapılandırma.

• Veri aktarımı sırasında Message Queue for Apache Kafka mesajları UTF-8 dizeleri olarak seri hale getirir. İkili veri biçimleri desteklenmez.

• AnalyticDB sink connector özel bir uç nokta kullanıyorsa, Function Compute çalışma zamanı ortamı varsayılan olarak buna erişemez. Ağ bağlantısını etkinleştirmek için Function Compute konsolundaki işlev hizmetini, hedef AnalyticDB örneğinizle aynı VPC ve vSwitch ile yapılandırın. Daha fazla bilgi için bkz. Hizmet güncelleme.

• Bir connector oluşturduğunuzda Message Queue for Apache Kafka otomatik olarak bir hizmet bağlantılı rol oluşturur.

  • Hizmet bağlantılı bir rol yoksa Message Queue for Apache Kafka otomatik olarak bir tane oluşturur. Bu rol, Message Queue for Apache Kafka hizmetine veri aktarmak için gereken izinleri verir.

  • Hizmet bağlantılı bir rol zaten varsa Message Queue for Apache Kafka başka bir rol oluşturmaz.

  Hizmet bağlantılı roller hakkında daha fazla bilgi için bkz. Hizmet bağlantılı roller.

Yordam

Bir AnalyticDB sink connector kullanarak Message Queue for Apache Kafka örneğinin bir kaynak konusundan bir AnalyticDB veritabanına veri aktarmak için şu adımları izleyin:

1. İsteğe bağlı: AnalyticDB Sink Connector için gerekli konuyu ve grubu oluşturun

   Konuyu ve Grubu özelleştirmeniz gerekmiyorsa bu adımı atlayıp sonraki adımda bunların otomatik olarak oluşturulmasını seçebilirsiniz.

   Önemli

   AnalyticDB sink connector tarafından gereken bazı konuların local storage kullanması gerekir. Ana sürüm 0.10.2 çalıştıran Message Queue for Apache Kafka örneklerinde local storage kullanan konuları manuel olarak oluşturamazsınız. Bu konular yalnızca otomatik olarak oluşturulabilir.

   1. AnalyticDB sink connector tarafından gereken konuları oluşturma

   2. AnalyticDB sink connector tarafından gereken grubu oluşturma

2. AnalyticDB sink connector oluşturma ve dağıtma

3. Hizmetleri yapılandırma

   1. Function Compute hizmetini yapılandırma

   2. Hedef AnalyticDB veritabanını yapılandırma

4. Sonucu doğrulama

   1. Test mesajları gönderme

   2. Sonucu doğrulama

Gerekli konuları oluşturma

AnalyticDB sink connector tarafından gereken beş konuyu Message Queue for Apache Kafka konsolunda manuel olarak oluşturabilirsiniz: bir görev ofset konusu, bir görev yapılandırma konusu, bir görev durumu konusu, bir dead-letter queue topic ve bir error data topic. Bu konuların bölüm sayısı ve depolama motoru gereksinimleri farklıdır. Daha fazla bilgi için Kaynak hizmeti yapılandırma parametreleri tablosuna bakın.

1. ApsaraMQ for Kafka konsolunda oturum açın.

2. Overview sayfasında Resource Distribution bölümünden bir bölge seçin.

   Önemli

   Konuları, uygulamanızın bulunduğu bölgede (ECS örneğinin dağıtıldığı bölgede) oluşturmanız gerekir. Konular bölgeler arasında kullanılamaz. Örneğin, Çin (Pekin) bölgesinde bir konu oluşturulursa mesaj üreticisi ve tüketici de Çin (Pekin) bölgesindeki bir ECS örneğinde çalışmalıdır.

3. Instances sayfasında hedef örneğin adına tıklayın.

4. Sol gezinme bölmesinde Topics öğesine tıklayın.

5. Topics sayfasında Create Topic öğesine tıklayın.

6. Create Topic panelinde konu özelliklerini ayarlayın ve ardından OK öğesine tıklayın.

   Parametre	Açıklama	Örnek
   Name	Konunun adı. Not Kafka'da xxx_xxx ve xxx.xxx gibi konu adları aynı kabul edilir. Yinelenen bir ada sahip bir konu oluşturmaya çalışırsanız hata bildirilir.	demo
   Description	Konuyla ilgili kısa bir açıklama.	demo test
   Partitions	Konudaki bölüm sayısı.	12
   Storage Engine Not Şu anda yalnızca Serverless olmayan Professional Edition örnekler için bir depolama motoru türü seçebilirsiniz. Diğer örneklerde bu parametre desteklenmez ve varsayılan olarak Cloud Storage olarak ayarlanır.	Konu mesajlarının depolama motoru. ApsaraMQ for Kafka aşağıdaki iki depolama motorunu destekler. Cloud Storage: Alt katman Alibaba Cloud disklerini kullanır. Bu depolama motoru düşük gecikme süresi, yüksek performans, yüksek dayanıklılık ve yüksek güvenilirlik sunar. Dağıtılmış üç çoğaltma mekanizması kullanır. Örneğin Instance Edition değeri Standard (High Write) ise depolama motoru yalnızca Cloud Storage olabilir. Local Storage: Yerel Kafka eşzamanlı çoğaltma (ISR) çoğaltma algoritmasını ve dağıtılmış üç çoğaltma mekanizmasını kullanır.	Cloud Storage
   Message Type	Konu mesajlarının türü. Normal Message: Varsayılan olarak aynı anahtara sahip mesajlar aynı bölüme dağıtılır ve bir bölüm içindeki mesajlar gönderildikleri sırayla saklanır. Kümedeki bir makine arızalanırsa mesajlar sırasız hale gelebilir. Storage Engine değerini Cloud Storage olarak ayarladığınızda varsayılan olarak Normal Message seçilir. Partitionally Ordered Message: Varsayılan olarak aynı anahtara sahip mesajlar aynı bölüme dağıtılır ve bir bölüm içindeki mesajlar gönderildikleri sırayla saklanır. Kümedeki bir makine arızalanırsa bile bölüm içindeki mesajların sırası korunur. Ancak bazı bölümlere mesaj gönderme işlemi başarısız olabilir. Bölümler kurtarıldıktan sonra normal işlemler devam eder. Storage Engine değerini Local Storage olarak ayarladığınızda varsayılan olarak Partitionally Ordered Message seçilir.	Normal Message
   Log Cleanup Policy	Konu günlüklerinin temizleme ilkesi. Storage Engine değerini Local Storage olarak ayarladığınızda Log Cleanup Policy değerini yapılandırmanız gerekir. Depolama Motorunu Local Storage olarak yalnızca Professional Edition örnekler için ayarlayabilirsiniz. Standard Edition örnekler desteklenmez. ApsaraMQ for Kafka aşağıdaki iki temizleme ilkesini destekler. Delete: Varsayılan mesaj temizleme ilkesi. Disk kapasitesi yeterliyse mesajlar maksimum saklama süresi boyunca korunur. Disk kapasitesi yetersizse (genellikle disk kullanımı %85'i aştığında) hizmet kullanılabilirliğini sağlamak için eski mesajlar erken silinir. Compact: Kafka Log Compaction temizleme ilkesini kullanır. Log Compaction ilkesi aynı anahtara sahip mesajlar için en son değerin her zaman korunmasını sağlar. Bu durum başlıca sistem arızasından sonra durumun kurtarılması veya sistem yeniden başlatıldıktan sonra önbelleğin yeniden yüklenmesi gibi senaryolar için geçerlidir. Örneğin Kafka Connect veya Confluent Schema Registry kullanırken sistem durumunu veya yapılandırma bilgilerini saklamak için bir Kafka Compact Topic kullanmanız gerekir. Önemli Compact konular genellikle yalnızca Kafka Connect veya Confluent Schema Registry gibi belirli ekosistem bileşenlerinde kullanılır. Diğer mesaj gönderme ve alma senaryolarındaki konular için bu özelliği ayarlamayın. Daha fazla bilgi için bkz. ApsaraMQ for Kafka Demo Kitaplığı.	Compact
   Tag	Konunun etiketleri.	demo

   Konu oluşturulduktan sonra Topics sayfasındaki listede görüntülenir.

AnalyticDB Sink Connector için gereken grubu oluşturma

AnalyticDB Sink Connector veri senkronizasyon görevi için grubuMessage Queue for Apache Kafka konsolunda manuel olarak oluşturabilirsiniz. Grubun adı connect-Görev Adı biçiminde olmalıdır. Daha fazla bilgi için bkz. Kaynak hizmeti parametreleri.

1. ApsaraMQ for Kafka konsolunda oturum açın.

2. Overview sayfasında Resource Distribution bölümünden bir bölge seçin.

3. Instances sayfasında hedef örneğin adına tıklayın.

4. Sol gezinme bölmesinde Groups öğesine tıklayın.

5. Groups sayfasında Create Group öğesine tıklayın.

6. Create Group panelinde Group ID metin kutusuna grup için bir ad girin, Description metin kutusuna kısa bir açıklama girin, gruba etiketler ekleyin ve ardından OK öğesine tıklayın.

   Grup oluşturulduktan sonra Groups sayfasındaki listede görüntülenir.

AnalyticDB sink connector oluşturma ve dağıtma

1. ApsaraMQ for Kafka konsolunda oturum açın.

2. Overview sayfasında Resource Distribution bölümünden bir bölge seçin.

3. Sol gezinme bölmesinde Connectors öğesine tıklayın.

4. Connectors sayfasında Select Instance açılır listesinden connector'ın ait olduğu örneği seçin ve Create Connector öğesine tıklayın.

5. Create Connector sihirbazında aşağıdaki adımları tamamlayın.

   1. Configure Basic Information sekmesinde parametreleri gerektiği gibi yapılandırın ve ardından Next öğesine tıklayın.

      Parametre	Açıklama	Örnek değer
      Name	Connector'ın adı. Adlandırma kuralları: Ad 1 ila 48 karakter uzunluğunda olabilir. İlk karakter büyük harf, küçük harf veya rakam olabilir. Kalan karakterler küçük harf, rakam veya tire (-) olabilir. Ad aynı Message Queue for Apache Kafka örneği içinde benzersiz olmalıdır. Bir Connector için veri senkronizasyon görevi connect-görev adı adlı bir Group kullanmalıdır. Group'u manuel olarak oluşturmazsanız sistem sizin için otomatik olarak oluşturur.	kafka-adb-sink
      Instance	Örnek adı ve Kimliği otomatik olarak doldurulur.	demo alikafka_post-cn-st21p8vj****

   2. Configure Source Service sekmesinde Data Source olarak Message Queue for Apache Kafka seçeneğini belirleyin, aşağıdaki parametreleri yapılandırın ve ardından Next öğesine tıklayın.

      Tablo 1. Kaynak hizmeti parametreleri

      Parametre	Açıklama	Örnek
      Data Source Topic	Verilerin senkronize edileceği konu.	adb-test-input
      Consumer Thread Concurrency	Kaynak konu için eşzamanlı tüketici iş parçacığı sayısı. Varsayılan değer: 6. Geçerli değerler: 1 2 3 6 12	6
      Consumer Offset	Tüketimin başlayacağı konum. Geçerli değerler: Earliest Offset: Tüketimi en eski kullanılabilir ofsetten başlatır. Latest Offset: Tüketimi en son kullanılabilir ofsetten başlatır.	Earliest Offset
      VPC ID	Veri senkronizasyon görevinin bulunduğu VPC. Bu parametre Configure Runtime Environment öğesine tıkladıktan sonra görüntülenir. Bu varsayılan olarak Message Queue for Apache Kafka örneğinin VPC'sidir ve manuel giriş gerektirmez.	vpc-bp1xpdnd3l***
      vSwitch ID	Veri senkronizasyon görevinin bulunduğu vSwitch. Bu parametre Configure Runtime Environment öğesine tıkladıktan sonra görüntülenir. Bu vSwitch Message Queue for Apache Kafka örneğiyle aynı VPC'de bulunmalıdır. Bu varsayılan olarak Message Queue for Apache Kafka örneğini dağıttığınızda belirtilen vSwitch'tir.	vsw-bp1d2jgg81***
      Failure Handling Policy	Mesaj gönderme hatasının gerçekleştiği bölümlerin işlenmesine yönelik ilke. Bu parametre Configure Runtime Environment öğesine tıkladıktan sonra görüntülenir. Geçerli değerler: Continue Subscription: Aboneliği sürdürür ve bir hata günlüğü yazdırır. Stop Subscription: Aboneliği durdurur ve bir hata günlüğü yazdırır. Not Günlüklerin nasıl görüntüleneceğini öğrenmek için bkz. Connector ile ilgili işlemler. Hata kodlarına göre çözümler bulmak için bkz. Hata kodları.	Continue Subscription
      Resource Creation Method	Connector tarafından gereken konuların ve grubun oluşturulma yöntemi. Bu parametre Configure Runtime Environment öğesine tıkladıktan sonra görüntülenir. Auto Manual	Auto
      Connector Consumer Group	Connector'ın veri senkronizasyon görevi tarafından kullanılan grup. Bu parametreyi görüntülemek için Configure Runtime Environment öğesine tıklayın. Grubun adı connect-görev adı biçiminde olmalıdır.	connect-kafka-adb-sink
      Task Offset Topic	Tüketici ofsetlerini saklamak için kullanılan konu. Bu parametre Configure Runtime Environment öğesine tıkladıktan sonra görüntülenir. Konu: Adın connect-offset ile başlaması önerilir. Bölümler: Bölüm sayısı 1'den büyük olmalıdır. Depolama Motoru: Konu local storage kullanmalıdır. cleanup.policy: Konunun günlük temizleme ilkesi compact olmalıdır.	connect-offset-kafka-adb-sink
      Task Configuration Topic	Görev yapılandırmalarını saklamak için kullanılan konu. Bu parametre Configure Runtime Environment öğesine tıkladıktan sonra görüntülenir. Konu: Adın connect-config ile başlaması önerilir. Bölümler: Bölüm sayısı 1 olmalıdır. Depolama Motoru: Konu local storage kullanmalıdır. cleanup.policy: Konunun günlük temizleme ilkesi compact olmalıdır.	connect-config-kafka-adb-sink
      Task Status Topic	Görev durumunu saklamak için kullanılan konu. Bu parametre Configure Runtime Environment öğesine tıkladıktan sonra görüntülenir. Konu: Adın connect-status ile başlaması önerilir. Bölümler: Önerilen bölüm sayısı 6'dır. Depolama Motoru: Konu local storage kullanmalıdır. cleanup.policy: Konunun günlük temizleme ilkesi compact olmalıdır.	connect-status-kafka-adb-sink
      Dead-letter Queue Topic	Connect framework'ündan gelen hata verilerini saklamak için kullanılan konu. Bu parametreyi görüntülemek için Configure Runtime Environment öğesine tıklayın. Bu konu, konu kaynaklarını tasarruf etmek amacıyla error data topic ile aynı olabilir. Konu: Adın connect-error ile başlaması önerilir. Bölümler: Önerilen bölüm sayısı 6'dır. Depolama Motoru: Konu local storage veya cloud storage kullanabilir.	connect-error-kafka-adb-sink
      Error Data Topic	Sink'ten gelen özel durum verilerini saklamak için kullanılan konu. Bu parametreyi görüntülemek için Configure Runtime Environment öğesine tıklayın. Bu konu, konu kaynaklarını tasarruf etmek amacıyla dead-letter queue topic ile aynı olabilir. Konu: Adın connect-error ile başlaması önerilir. Bölümler: Önerilen bölüm sayısı 6'dır. Depolama Motoru: Konu local storage veya cloud storage kullanabilir.	connect-error-kafka-adb-sink

   3. Configure Destination Service sekmesinde Destination Service olarak AnalyticDB seçeneğini belirleyin, aşağıdaki parametreleri yapılandırın ve ardından Create öğesine tıklayın.

      Parametre	Açıklama	Örnek
      Instance Type	Hedef AnalyticDB örneğinin türü. Geçerli değerler: AnalyticDB for MySQL ve AnalyticDB for PostgreSQL.	MySQL
      AnalyticDB Instance ID	Hedef AnalyticDB for MySQL veya AnalyticDB for PostgreSQL örneğinin kimliği.	am-bp139yqk8u1ik****
      Database Name	AnalyticDB örneğindeki hedef veritabanının adı.	adb_demo
      Table Name	Mesajları saklamak için AnalyticDB veritabanındaki hedef tablonun adı.	user
      Database Username	Verileri içe aktarmak için AnalyticDB örneğine bağlanmak üzere kullanılan kullanıcı adı.	adbmysql
      Database Password	Verileri içe aktarmak için AnalyticDB örneğine bağlanmak üzere kullanılan parola. Parolayı örnek oluştururken ayarlarsınız. Parolayı unutursanız sıfırlayabilirsiniz. AnalyticDB for MySQL: Parolanın nasıl sıfırlanacağı hakkında bilgi için bkz. Hesap parolasını değiştirme. AnalyticDB for PostgreSQL: AnalyticDB for PostgreSQL konsolunda oturum açın, örnek adına tıklayın ve sol gezinme bölmesinde Account Management öğesine tıklayın. Parolasını sıfırlamak istediğiniz hesabı bulun ve Actions sütununda Reset Password öğesine tıklayın.	****

      Not

      Message Queue for Apache Kafka görevi oluştururken kullanıcı adını ve parolayı Function Compute işlevine ortam değişkenleri olarak iletir. Görev oluşturulduktan sonra Message Queue for Apache Kafka bu bilgileri kaydetmez.

      Connector oluşturulduktan sonra Connectors sayfasında görüntüleyebilirsiniz.

6. Connectors sayfasında oluşturduğunuz connector'ı bulun ve Actions sütununda Deploy öğesine tıklayın.

Function Compute hizmetini yapılandırma

Message Queue for Apache Kafka konsolunda bir AnalyticDB Sink Connector başarıyla oluşturup dağıttıktan sonra Function Compute connector için otomatik olarak bir hizmet ve bir işlev oluşturur. Hizmet kafka-service-<connector_adı>-<random_string> biçiminde adlandırılır ve işlev fc-adb-<random_string> biçiminde adlandırılır.

1. Connectors sayfasında hedef connector'ı bulun ve Actions sütununda Configure Function öğesine tıklayın.

   Bu işlem sizi Function Compute konsoluna yönlendirir.

2. Function Compute konsolunda otomatik olarak oluşturulan işlev hizmetini bulun ve VPC ile vSwitch bilgilerini yapılandırın. Belirli adımlar için bkz. Hizmet güncelleme.

Hedef AnalyticDB veritabanını yapılandırma

Function Compute hizmetini yapılandırdıktan sonra vSwitch'inin CIDR bloğunu AnalyticDB örneğinizin izin listesine eklemeniz gerekir. Bu CIDR bloğunu VPC konsolununvSwitch sayfasında Function Compute hizmetinin VPC ve vSwitch'ine karşılık gelen satırda bulabilirsiniz.

• AnalyticDB for MySQL: AnalyticDB for MySQL konsolunda oturum açın ve izin listesini yapılandırın. Daha fazla bilgi için bkz. IP adresi izin listesi yapılandırma.

• AnalyticDB for PostgreSQL: AnalyticDB for PostgreSQL konsolunda oturum açın ve izin listesini yapılandırın. Daha fazla bilgi için bkz. İzin listesi yapılandırma.

Test mesajları gönderme

1. Connectors sayfasında hedef connector'ı bulun ve Actions sütununda Test öğesine tıklayın.

2. Send Message panelinde bir test mesajı gönderin.

   • Sending Method değerini Console olarak ayarlayın.

     1. Message Key metin kutusuna mesajın anahtarını girin. Örneğin demo.

     2. Message Content metin kutusuna test mesajının içeriğini girin. Örneğin {"key": "test"}.

     3. Mesajın belirli bir bölüme gönderilip gönderilmeyeceğini belirtmek için Send to Specified Partition değerini ayarlayın.

        • Yes öğesine tıklayın ve Partition ID metin kutusuna bölüm kimliğini girin. Örneğin 0. Bölüm kimliğini sorgulamak için bkz. Bölüm durumunu görüntüleme.

        • Bir bölüm belirtmemek için No öğesine tıklayın.

   • Sending Method değerini Docker olarak ayarlayın. Mesajı göndermek için Run the Docker container to produce a sample message bölümündeki Docker komutunu çalıştırın.

   • Sending Method değerini SDK olarak ayarlayın. Mesaj göndermek için gerekli dil veya çerçeve için bir SDK ve bir bağlantı türü seçin.

Sonucu doğrulama

Message Queue for Apache Kafka içindeki kaynak konuya mesaj gönderdikten sonra AnalyticDB for MySQL konsolunda veya AnalyticDB for PostgreSQL konsolunda oturum açın, veritabanına bağlanın ve Data Management Service (DMS) penceresinin SQL Window bölümüne gidin. Verilerin doğru şekilde aktarılıp aktarılmadığını kontrol etmek için ilgili örnekteki tabloyu sorgulayın.