ApsaraMQ for Kafka：Buat konektor sumber MySQL

Tindakan pencegahan

• Wilayah
  • Sumber data dan tujuan instance Message Queue for Apache Kafka Anda mungkin tidak ditempatkan di wilayah yang sama. Dalam kasus ini, pastikan bahwa Anda memiliki instance Cloud Enterprise Network (CEN) dalam akun Alibaba Cloud Anda dan bahwa VPC dari sumber data dan instance Message Queue for Apache Kafka Anda terhubung ke instance CEN. Selain itu, pastikan bahwa bandwidth untuk koneksi lintas wilayah dikonfigurasi untuk memastikan konektivitas ujung ke ujung antara sumber data dan tujuan data.

    Jika tidak, instance CEN mungkin secara otomatis dibuat, dan VPC dari instance Message Queue for Apache Kafka tujuan serta instance Elastic Compute Service (ECS) tempat grup sumber daya eksklusif Anda berada terhubung ke instance CEN untuk memastikan konektivitas ujung ke ujung. Bandwidth dari instance CEN yang dibuat secara otomatis sangat rendah karena bandwidth tidak dikonfigurasi secara manual. Ini dapat menyebabkan kesalahan konektivitas saat Anda membuat konektor sumber MySQL atau saat konektor tersebut berjalan.

  • Sumber data dan tujuan instance Message Queue for Apache Kafka Anda mungkin berada di wilayah yang sama. Dalam kasus ini, saat Anda membuat konektor sumber MySQL, ENI secara otomatis dibuat dalam VPC dari sumber data atau instance Message Queue for Apache Kafka. ENI juga secara otomatis terikat ke instance ECS tempat grup sumber daya eksklusif Anda berada. Ini memastikan konektivitas ujung ke ujung antara sumber data dan tujuan data.
• Grup Sumber Daya Eksklusif dalam DataWorks
  • DataWorks memungkinkan Anda menggunakan setiap grup sumber daya eksklusif untuk menjalankan hingga tiga konektor sumber MySQL. Jika DataWorks menemukan bahwa grup sumber daya yang ada telah digunakan untuk menjalankan kurang dari tiga konektor sumber MySQL saat Anda membuat konektor sumber MySQL, DataWorks menggunakan grup sumber daya ini untuk menjalankan konektor sumber MySQL yang baru dibuat.
  • Setiap grup sumber daya eksklusif dalam DataWorks dapat dikaitkan dengan ENI dari maksimal dua VPC. DataWorks mungkin gagal membuat konektor sumber MySQL menggunakan grup sumber daya jika ENI yang akan dikaitkan dengan grup sumber daya dan ENI yang sudah dikaitkan dengan grup sumber daya memiliki blok CIDR yang tumpang tindih atau karena batasan teknis lainnya. Dalam kasus ini, meskipun kurang dari tiga konektor berjalan pada grup sumber daya ini, DataWorks tetap membuat grup sumber daya untuk memastikan bahwa konektor sumber MySQL dapat dibuat.

Buat dan sebarkan konektor sumber MySQL

1. Masuk ke konsol ApsaraMQ for Kafka.

2. Di bagian Resource Distribution dari halaman Overview, pilih wilayah tempat instance ApsaraMQ for Kafka yang ingin Anda kelola berada.

3. Di panel navigasi di sebelah kiri, klik Connectors.

4. Di halaman Connectors, pilih instance tempat konektor milik dari daftar drop-down Select Instance dan klik Create Connector.

5. Di wizard Create Connector, lakukan operasi berikut:
   1. Di langkah Configure Basic Information, masukkan nama konektor di bidang Name dan klik Next.

      Parameter	Deskripsi	Contoh
      Name	Nama konektor. Tentukan nama konektor berdasarkan konvensi penamaan berikut: Nama konektor harus memiliki panjang 1 hingga 48 karakter dan dapat berisi digit, huruf kecil, dan tanda hubung (-). Nama tidak boleh dimulai dengan tanda hubung (-). Setiap nama konektor harus unik dalam sebuah instance ApsaraMQ for Kafka. Konektor harus menggunakan grup konsumen yang dinamai dalam format connect-Nama konektor.Group Jika Anda belum membuat grup konsumen ini, Message Queue for Apache Kafka secara otomatis membuat grup konsumen untuk Anda.Group	kafka-source-mysql
      Instance	Informasi tentang instance Message Queue for Apache Kafka. Secara default, nama dan ID instance ditampilkan.	demo alikafka_post-cn-st21p8vj****

   2. Di langkah Configure Source Service, pilih ApsaraDB RDS for MySQL sebagai layanan sumber, konfigurasikan parameter yang dijelaskan dalam tabel berikut, lalu klik Next.

      Parameter	Deskripsi	Contoh
      Region of ApsaraDB RDS for MySQL Instance	ID wilayah instance ApsaraDB RDS for MySQL. Pilih ID wilayah dari daftar drop-down.	China (Shenzhen)
      ApsaraDB RDS for MySQL Instance ID	ID instance ApsaraDB RDS for MySQL dari mana Anda ingin menyinkronkan data.	rm-wz91w3vk6owmz****
      Database Name	Nama database ApsaraDB RDS for MySQL dari mana Anda ingin menyinkronkan data.	mysql-to-kafka
      Database Account	Nama pengguna akun yang digunakan untuk mengakses database ApsaraDB RDS for MySQL.	mysql_to_kafka
      Password of Database Account	Kata sandi akun yang digunakan untuk mengakses database ApsaraDB RDS for MySQL.	N/A
      Database Table	Nama satu atau lebih tabel ApsaraDB RDS for MySQL dari mana Anda ingin menyinkronkan data. Pisahkan beberapa nama tabel dengan koma (,). Tabel sumber dan topik tujuan saling berkaitan berdasarkan pemetaan satu-satu.	mysql_tbl
      Tables to Automatically Add	Ekspresi reguler berdasarkan sistem dapat secara otomatis menambahkan tabel lain dalam database. Saat sistem mendeteksi bahwa tabel dibuat dalam database dan sesuai dengan ekspresi reguler, sistem secara otomatis menyinkronkan data dalam tabel ke Message Queue for Apache Kafka. Anda dapat menggunakan .* untuk menentukan semua tabel dalam database.	.*
      Topic Prefix	Awalan yang digunakan untuk menamai topik yang dibuat secara otomatis untuk konektor dalam ApsaraMQ for Kafka. Setiap nama topik terdiri dari awalan dan nama tabel sumber yang sesuai dalam database ApsaraDB RDS for MySQL. Pastikan bahwa awalan bersifat unik secara global.	mysql

      Penting

      Pastikan bahwa akun database ApsaraDB RDS for MySQL diberikan setidaknya izin berikut:

      • SELECT
      • REPLICATION SLAVE
      • REPLICATION CLIENT

      Pernyataan berikut memberikan contoh tentang cara memberikan izin kepada akun:

      GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'Nama akun' @'%'; // Berikan izin SELECT, REPLICATION SLAVE, dan REPLICATION CLIENT kepada akun database. 

   3. Di langkah Configure Destination Service, tujuan instance ApsaraMQ for Kafka tempat Anda ingin menyinkronkan data muncul. Konfirmasikan informasi dan klik Create.
6. Pergi ke halaman Connectors, temukan konektor yang Anda buat, lalu klik Deploy di kolom Actions.
   Di halaman Connectors, jika nilai parameter Status ditampilkan sebagai Running untuk konektor, konektor telah dibuat.

   Catatan Jika konektor gagal dibuat, periksa apakah semua prasyarat yang dijelaskan dalam topik ini telah dipenuhi.

   Jika Anda perlu memodifikasi konfigurasi konektor sumber MySQL, klik Task Configurations di kolom Actions. Di konsol DataWorks yang Anda dialihkan, modifikasi konfigurasi konektor.

Verifikasi hasil

1. Masukkan data ke dalam tabel sumber dalam database ApsaraDB RDS for MySQL.
   Kode sampel berikut memberikan contoh tentang cara memasukkan data ke dalam tabel sumber:

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

   Untuk informasi lebih lanjut tentang pernyataan SQL, lihat Pernyataan SQL yang Sering Digunakan untuk MySQL.
2. Gunakan fitur kueri pesan ApsaraMQ for Kafka untuk memeriksa apakah data tabel sumber dalam database ApsaraDB RDS for MySQL disinkronkan ke topik yang sesuai dalam instance ApsaraMQ for Kafka Anda.
   Untuk informasi lebih lanjut, lihat Kueri Pesan.
   Kode sampel berikut memberikan contoh data yang disinkronkan dari tabel sumber dalam ApsaraDB RDS for MySQL ke topik dalam ApsaraMQ for Kafka. Untuk informasi lebih lanjut tentang struktur pesan dan bidang, lihat Lampiran: Format Pesan.

   {
       "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"
   }