ApsaraMQ for Kafka:Create a MySQL source connector

Çalışma mantığı

ApsaraMQ for Kafka konsolu üzerinden bir MySQL kaynak bağlayıcısı oluşturduğunuzda sistem aşağıdaki işlemleri otomatik olarak gerçekleştirir:

1. DataWorks Basic Edition'ı (ücretsiz) etkinleştirir.

2. Bir DataWorks çalışma alanı ve veri entegrasyonu için özel bir kaynak grubu oluşturur.

3. Kafka örneğinizde hedef konular oluşturur; her kaynak tablo için bir konu oluşturulur.

Özel kaynak grubu (4 vCPU, 8 GB bellek), veri senkronizasyon görevlerini çalıştırır. Kaynak grupları aylık abonelik modeliyle sunulur ve süresi dolduğunda otomatik olarak yenilenir.

<topic-prefix>_<source-table-name>

Ön koşullar

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

• Bağlayıcı özelliği etkin bir ApsaraMQ for Kafka örneği. Örnek şu bölgelerden birinde dağıtılmalıdır: Çin (Shenzhen), Çin (Chengdu), Çin (Pekin), Çin (Zhangjiakou), Çin (Hangzhou), Çin (Şanghay) veya Singapur.

• Bir ApsaraDB RDS for MySQL örneği, bir veritabanı, veritabanı hesabı ve en az bir tablo. SQL referansı için bkz. MySQL için sık kullanılan SQL deyimleri

• RDS MySQL örneğinizin VPC'si ile Kafka örneğinizin VPC'si arasında örtüşmeyen CIDR blokları

• DataWorks'ün ENI'lerinize erişmesi için Cloud Resource Access Authorization sayfasında yetkilendirme yapılması

• Hem veri kaynağının (ApsaraDB RDS for MySQL örneği) hem de veri hedefinin (ApsaraMQ for Kafka örneği) aynı Alibaba Cloud hesabı altında oluşturulmuş olması

GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO '<your-username>'@'%';

Bağlayıcıyı oluşturma ve dağıtma

1. ApsaraMQ for Kafka konsoluna giriş yapın.

2. Overview sayfasının Resource Distribution bölümünde örneğinizin bulunduğu bölgeyi seçin.

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

4. Connectors sayfasında Select Instance açılır listesinden örneğinizi seçin ve Create Connector düğmesine tıklayın.

5. Üç adımlı sihirbazı tamamlayın:

   Step 1: Configure basic information

   Parametre	Açıklama	Örnek
   Name	Örnek içinde benzersiz bir bağlayıcı adı. 1–48 karakter; rakamlar, küçük harfler ve tire (-) karakterlerine izin verilir. Tire ile başlayamaz. Sistem otomatik olarak connect-<connector-name> adlı bir tüketici grubu oluşturur.	kafka-source-mysql
   Instance	Seçilen Kafka örneğinin adını ve kimliğini görüntüler.	demo alikafka_post-cn-st21p8vj****

   Step 2: Configure source service Kaynak hizmet olarak ApsaraDB RDS for MySQL seçeneğini belirleyin ve aşağıdaki parametreleri yapılandırın:

   Parametre	Açıklama	Örnek
   Region of ApsaraDB RDS for MySQL Instance	Kaynak RDS MySQL örneğinin bulunduğu bölge.	China (Shenzhen)
   ApsaraDB RDS for MySQL Instance ID	Kaynak veritabanının örnek kimliği.	rm-wz91w3vk6owmz****
   Database Name	Senkronize edilecek veritabanının adı.	mysql-to-kafka
   Database Account	Veritabanı bağlantısı için kullanıcı adı.	mysql_to_kafka
   Password of Database Account	Veritabanı bağlantısı için parola.	--
   Database Table	Virgülle (,) ayrılmış bir veya daha fazla tablo adı. Her tablo bir konuyla eşleşir.	mysql_tbl
   Tables to Automatically Add	Yeni tabloları otomatik olarak algılayıp senkronize etmek için kullanılan normal ifade. Tüm tablolar için .* kullanın.	.*
   Topic Prefix	Otomatik oluşturulan konu adları için ön ek. Genel olarak benzersiz olmalıdır.	mysql

   Step 3: Configure destination service Hedef Kafka örneğini onaylayın ve Create düğmesine tıklayın. Adımlar arasında ilerlemek için Next ve Next düğmelerine tıklayın.

6. Connectors sayfasında bağlayıcıyı bulun ve Actions sütunundaki Deploy düğmesine tıklayın. Status sütununda Running durumu göründüğünde bağlayıcı etkin hale gelir ve verileri senkronize etmeye başlar.

Veri senkronizasyonunu doğrulama

1. Kaynak tabloya bir test satırı ekleyin. Daha fazla SQL örneği için bkz. MySQL için sık kullanılan SQL deyimleri.

      INSERT INTO mysql_tbl
          (mysql_title, mysql_author, submission_date)
          VALUES
          ("mysql2kafka", "tester", NOW());

2. ApsaraMQ for Kafka'daki mesaj sorgulama özelliğini kullanarak verilerin ilgili konuya ulaştığını onaylayın. Talimatlar için bkz. Mesajları sorgulama. Başarıyla senkronize edilen bir INSERT olayı aşağıdaki gibi görünür:

      {
          "schema": {
              "dataColumn": [
                  {
                      "name": "mysql_id",
                      "type": "LONG"
                  },
                  {
                      "name": "mysql_title",
                      "type": "STRING"
                  },
                  {
                      "name": "mysql_author",
                      "type": "STRING"
                  },
                  {
                      "name": "submission_date",
                      "type": "DATE"
                  }
              ],
              "primaryKey": [
                  "mysql_id"
              ],
              "source": {
                  "dbType": "MySQL",
                  "dbName": "mysql_to_kafka",
                  "tableName": "mysql_tbl"
              }
          },
          "payload": {
              "before": null,
              "after": {
                  "dataColumn": {
                      "mysql_title": "mysql2kafka",
                      "mysql_author": "tester",
                      "submission_date": 1614700800000
                  }
              },
              "sequenceId": "1614748790461000000",
              "timestamp": {
                  "eventTime": 1614748870000,
                  "systemTime": 1614748870925,
                  "checkpointTime": 1614748870000
              },
              "op": "INSERT",
              "ddl": null
          },
          "version": "0.0.1"
      }

Sonraki adımlar

• Kafka konularınızdaki mesajları sorgulayın

• Ek örnekler için bağlayıcı özelliğini etkinleştirin