ApsaraMQ for Kafka:Buat konektor sink AnalyticDB

Catatan penggunaan

• Anda dapat mengekspor data dari topik sumber instans Message Queue for Apache Kafka ke AnalyticDB for MySQL atau AnalyticDB for PostgreSQL hanya jika semua layanan berada di wilayah yang sama. Untuk informasi selengkapnya mengenai batasan konektor, lihat Batasan.

• Fitur ini didukung oleh Function Compute. Function Compute menyediakan kuota gratis. Biaya dikenakan untuk penggunaan sumber daya yang melebihi kuota gratis. Untuk informasi selengkapnya, lihat Ikhtisar penagihan Function Compute.

• Anda dapat mengkueri log pemanggilan fungsi di Function Compute untuk memecahkan masalah. Untuk informasi selengkapnya, lihat Konfigurasikan logging.

• Selama ekspor data, Message Queue for Apache Kafka melakukan serialisasi pesan sebagai string UTF-8. Format data biner tidak didukung.

• Jika konektor sink AnalyticDB menggunakan titik akhir pribadi, lingkungan runtime Function Compute tidak dapat mengaksesnya secara default. Untuk mengaktifkan konektivitas jaringan, konfigurasikan layanan fungsi di Konsol Function Compute dengan VPC dan vSwitch yang sama seperti instans AnalyticDB tujuan Anda. Untuk informasi selengkapnya, lihat Perbarui layanan.

• Message Queue for Apache Kafka secara otomatis membuat peran terkait layanan saat Anda membuat konektor.

  • Jika peran terkait layanan belum ada, Message Queue for Apache Kafka akan membuatnya secara otomatis. Peran ini memberikan izin yang diperlukan kepada Message Queue for Apache Kafka untuk mengekspor data.

  • Jika peran terkait layanan sudah ada, Message Queue for Apache Kafka tidak membuat peran baru.

  Untuk informasi selengkapnya mengenai peran terkait layanan, lihat Peran terkait layanan.

Prosedur

Untuk mengekspor data dari topik sumber instans Message Queue for Apache Kafka ke database AnalyticDB dengan konektor sink AnalyticDB, ikuti langkah-langkah berikut:

1. Opsi: Untuk Konektor Sink AnalyticDB, buat topik dan group yang diperlukan

   Jika Anda tidak perlu menyesuaikan Topik dan Group, Anda dapat melewati langkah ini dan memilih untuk membuatnya secara otomatis pada langkah berikutnya.

   Penting

   Beberapa topik yang diperlukan oleh konektor sink AnalyticDB harus menggunakan local storage. Untuk instans Message Queue for Apache Kafka yang menjalankan versi utama 0.10.2, Anda tidak dapat membuat topik yang menggunakan local storage secara manual. Topik tersebut hanya dapat dibuat secara otomatis.

   1. Buat topik yang diperlukan oleh konektor sink AnalyticDB

   2. Buat group yang diperlukan oleh konektor sink AnalyticDB

2. Buat dan deploy konektor sink AnalyticDB

3. Konfigurasikan layanan

   1. Konfigurasikan layanan Function Compute

   2. Konfigurasikan database AnalyticDB tujuan

4. Verifikasi hasil

   1. Kirim pesan uji

   2. Verifikasi hasil

Buat topik yang diperlukan

Anda dapat membuat lima topik yang diperlukan oleh konektor sink AnalyticDB secara manual di konsol Message Queue for Apache Kafka: topik offset tugas, topik konfigurasi tugas, topik status tugas, dead-letter queue topic, dan error data topic. Topik-topik ini memiliki persyaratan berbeda untuk jumlah partisi dan mesin penyimpanan. Untuk informasi selengkapnya, lihat tabel Parameter untuk mengonfigurasi layanan sumber.

1. Login ke ApsaraMQ for Kafka console.

2. Pada halaman Overview, pilih wilayah di bagian Resource Distribution.

   Penting

   Anda harus membuat topik di wilayah yang sama dengan aplikasi Anda, yaitu wilayah tempat Instance ECS dideploy. Topik tidak dapat digunakan lintas wilayah. Misalnya, jika topik dibuat di wilayah Tiongkok (Beijing), produsen dan konsumen pesan juga harus berjalan di Instance ECS di wilayah Tiongkok (Beijing).

3. Pada halaman Instances, klik nama instans target.

4. Di panel navigasi sebelah kiri, klik Topics.

5. Pada halaman Topics, klik Create Topic.

6. Di panel Create Topic, atur properti topik, lalu klik OK.

   Parameter	Deskripsi	Contoh
   Name	Nama topik. Catatan Di Kafka, nama topik seperti xxx_xxx dan xxx.xxx dianggap sama. Kesalahan akan dilaporkan jika Anda mencoba membuat topik dengan nama duplikat.	demo
   Description	Deskripsi singkat tentang topik.	demo test
   Partitions	Jumlah partisi dalam topik.	12
   Storage Engine Catatan Saat ini, Anda hanya dapat memilih jenis mesin penyimpanan untuk instans Edisi Profesional non-Serverless. Untuk instans lain, parameter ini tidak didukung dan diatur ke Cloud Storage secara default.	Mesin penyimpanan untuk pesan topik. ApsaraMQ for Kafka mendukung dua mesin penyimpanan berikut. Cloud Storage: Lapisan dasar menggunakan disk Alibaba Cloud. Mesin penyimpanan ini memiliki latensi rendah, performa tinggi, daya tahan tinggi, dan keandalan tinggi. Menggunakan mekanisme tiga replika terdistribusi. Jika Instance Edition instans adalah Standard (High Write), mesin penyimpanan hanya dapat berupa Cloud Storage. Local Storage: Menggunakan algoritma replikasi in-sync replica (ISR) Kafka asli dan mekanisme tiga replika terdistribusi.	Cloud Storage
   Message Type	Jenis pesan topik. Normal Message: Secara default, pesan dengan kunci yang sama didistribusikan ke partisi yang sama, dan pesan dalam satu partisi disimpan sesuai urutan pengiriman. Jika mesin di kluster mati, pesan mungkin menjadi tidak terurut. Saat Anda mengatur Storage Engine ke Cloud Storage, Normal Message dipilih secara default. Partitionally Ordered Message: Secara default, pesan dengan kunci yang sama didistribusikan ke partisi yang sama, dan pesan dalam satu partisi disimpan sesuai urutan pengiriman. Jika mesin di kluster mati, urutan pesan tetap dijamin dalam partisi tersebut. Namun, pengiriman pesan ke beberapa partisi mungkin gagal. Operasi normal dilanjutkan setelah partisi pulih. Saat Anda mengatur Storage Engine ke Local Storage, Partitionally Ordered Message dipilih secara default.	Normal Message
   Log Cleanup Policy	Kebijakan pembersihan log topik. Saat Anda mengatur Storage Engine ke Local Storage, Anda harus mengonfigurasi Log Cleanup Policy. Anda hanya dapat mengatur Storage Engine ke Local Storage untuk instans Edisi Profesional. Instans Edisi Standar tidak didukung. ApsaraMQ for Kafka mendukung dua kebijakan pembersihan berikut. Delete: Kebijakan pembersihan pesan default. Jika kapasitas disk mencukupi, pesan disimpan hingga periode retensi maksimum. Jika kapasitas disk tidak mencukupi (umumnya saat penggunaan disk melebihi 85%), pesan lama dihapus lebih awal untuk memastikan ketersediaan layanan. Compact: Menggunakan kebijakan pembersihan Kafka Log Compaction. Kebijakan Log Compaction memastikan bahwa untuk pesan dengan kunci yang sama, nilai terbaru selalu dipertahankan. Ini terutama berlaku untuk skenario seperti memulihkan status setelah sistem down atau memuat ulang cache setelah sistem restart. Misalnya, saat menggunakan Kafka Connect atau Confluent Schema Registry, Anda harus menggunakan Kafka Compact Topic untuk menyimpan status atau informasi konfigurasi sistem. Penting Topik compact umumnya hanya digunakan dalam komponen ekosistem tertentu, seperti Kafka Connect atau Confluent Schema Registry. Jangan atur properti ini untuk topik dalam skenario pengiriman dan penerimaan pesan lainnya. Untuk informasi selengkapnya, lihat ApsaraMQ for Kafka Demo Library.	Compact
   Tag	Tag topik.	demo

   Setelah topik dibuat, topik tersebut ditampilkan dalam daftar di halaman Topics.

Buat group yang diperlukan oleh Konektor Sink AnalyticDB

Anda dapat membuat group untuk tugas sinkronisasi data Konektor Sink AnalyticDB secara manual di konsol Message Queue for Apache Kafka. Nama group harus berupa connect-Nama Tugas. Untuk informasi selengkapnya, lihat Parameter layanan sumber.

1. Login ke ApsaraMQ for Kafka console.

2. Pada halaman Overview, pilih wilayah di bagian Resource Distribution.

3. Pada halaman Instances, klik nama instans target.

4. Di panel navigasi sebelah kiri, klik Groups.

5. Pada halaman Groups, klik Create Group.

6. Di panel Create Group, masukkan nama grup di kotak teks Group ID, masukkan deskripsi singkat di kotak teks Description, tambahkan tag ke grup, lalu klik OK.

   Setelah grup dibuat, grup tersebut ditampilkan dalam daftar di halaman Groups.

Buat dan deploy konektor sink AnalyticDB

1. Login ke ApsaraMQ for Kafka console.

2. Pada halaman Overview, pilih wilayah di bagian Resource Distribution.

3. Di panel navigasi sebelah kiri, klik Connectors.

4. Pada halaman Connectors, pilih instans tempat konektor berada dari daftar drop-down Select Instance dan klik Create Connector.

5. Di wizard Create Connector, lengkapi langkah-langkah berikut.

   1. Pada tab Configure Basic Information, konfigurasikan parameter sesuai kebutuhan, lalu klik Next.

      Parameter	Deskripsi	Nilai contoh
      Name	Nama konektor. Aturan penamaan: Nama dapat terdiri dari 1 hingga 48 karakter. Karakter pertama dapat berupa huruf kapital, huruf kecil, atau angka. Karakter berikutnya dapat berupa huruf kecil, angka, atau tanda hubung (-). Nama harus unik dalam instans Message Queue for Apache Kafka yang sama. Tugas sinkronisasi data untuk Konektor harus menggunakan Group bernama connect-nama tugas. Jika Anda tidak membuat Group secara manual, sistem akan membuatnya untuk Anda.	kafka-adb-sink
      Instance	Nama dan ID instans diisi secara otomatis.	demo alikafka_post-cn-st21p8vj****

   2. Pada tab Configure Source Service, pilih Message Queue for Apache Kafka sebagai Data Source, konfigurasikan parameter berikut, lalu klik Next.

      Tabel 1. Parameter layanan sumber

      Parameter	Deskripsi	Contoh
      Data Source Topic	Topik tempat data disinkronkan.	adb-test-input
      Consumer Thread Concurrency	Jumlah thread konsumen konkuren untuk topik sumber. Nilai default: 6. Nilai yang valid: 1 2 3 6 12	6
      Consumer Offset	Posisi mulai konsumsi. Nilai yang valid: Earliest Offset: Mulai konsumsi dari offset paling awal yang tersedia. Latest Offset: Mulai konsumsi dari offset paling baru yang tersedia.	Earliest Offset
      VPC ID	VPC tempat tugas sinkronisasi data berada. Parameter ini ditampilkan setelah Anda mengklik Configure Runtime Environment. Secara default, ini adalah VPC dari instans Message Queue for Apache Kafka dan tidak perlu dimasukkan secara manual.	vpc-bp1xpdnd3l***
      vSwitch ID	vSwitch tempat tugas sinkronisasi data berada. Parameter ini ditampilkan setelah Anda mengklik Configure Runtime Environment. vSwitch ini harus berada di VPC yang sama dengan instans Message Queue for Apache Kafka. Secara default, ini adalah vSwitch yang ditentukan saat Anda mendeploy instans Message Queue for Apache Kafka.	vsw-bp1d2jgg81***
      Failure Handling Policy	Kebijakan untuk menangani partisi tempat terjadi kegagalan pengiriman pesan. Parameter ini ditampilkan setelah Anda mengklik Configure Runtime Environment. Nilai yang valid: Continue Subscription: Melanjutkan berlangganan ke partisi dan mencetak log kesalahan. Stop Subscription: Menghentikan berlangganan ke partisi dan mencetak log kesalahan. Catatan Untuk mempelajari cara melihat log, lihat Operasi terkait konektor. Untuk menemukan solusi berdasarkan kode kesalahan, lihat Kode kesalahan.	Continue Subscription
      Resource Creation Method	Metode pembuatan topik dan grup yang diperlukan oleh konektor. Parameter ini ditampilkan setelah Anda mengklik Configure Runtime Environment. Auto Manual	Auto
      Connector Consumer Group	group yang digunakan oleh tugas sinkronisasi data konektor. Klik Configure Runtime Environment untuk menampilkan parameter ini. Nama group harus berupa connect-nama tugas.	connect-kafka-adb-sink
      Task Offset Topic	Topik yang digunakan untuk menyimpan offset konsumen. Parameter ini ditampilkan setelah Anda mengklik Configure Runtime Environment. Topik: Disarankan nama dimulai dengan connect-offset. Partisi: Jumlah partisi harus lebih dari 1. Mesin Penyimpanan: Topik harus menggunakan local storage. cleanup.policy: Kebijakan pembersihan log topik harus compact.	connect-offset-kafka-adb-sink
      Task Configuration Topic	Topik yang digunakan untuk menyimpan konfigurasi tugas. Parameter ini ditampilkan setelah Anda mengklik Configure Runtime Environment. Topik: Disarankan nama dimulai dengan connect-config. Partisi: Jumlah partisi harus 1. Mesin Penyimpanan: Topik harus menggunakan local storage. cleanup.policy: Kebijakan pembersihan log topik harus compact.	connect-config-kafka-adb-sink
      Task Status Topic	Topik yang digunakan untuk menyimpan status tugas. Parameter ini ditampilkan setelah Anda mengklik Configure Runtime Environment. Topik: Disarankan nama dimulai dengan connect-status. Partisi: Jumlah partisi yang disarankan adalah 6. Mesin Penyimpanan: Topik harus menggunakan local storage. cleanup.policy: Kebijakan pembersihan log topik harus compact.	connect-status-kafka-adb-sink
      Dead-letter Queue Topic	Topik yang digunakan untuk menyimpan data kesalahan dari framework Connect. Klik Configure Runtime Environment untuk menampilkan parameter ini. Topik ini dapat sama dengan error data topic untuk menghemat sumber daya topik. Topik: Disarankan nama dimulai dengan connect-error. Partisi: Jumlah partisi yang disarankan adalah 6. Mesin Penyimpanan: Topik dapat menggunakan local storage atau cloud storage.	connect-error-kafka-adb-sink
      Error Data Topic	Topik yang digunakan untuk menyimpan data pengecualian dari Sink. Klik Configure Runtime Environment untuk menampilkan parameter ini. Topik ini dapat sama dengan dead-letter queue topic untuk menghemat sumber daya topik. Topik: Disarankan nama dimulai dengan connect-error. Partisi: Jumlah partisi yang disarankan adalah 6. Mesin Penyimpanan: Topik dapat menggunakan local storage atau cloud storage.	connect-error-kafka-adb-sink

   3. Pada tab Configure Destination Service, pilih AnalyticDB sebagai Destination Service, konfigurasikan parameter berikut, lalu klik Create.

      Parameter	Deskripsi	Contoh
      Instance Type	Jenis instans AnalyticDB tujuan. Nilai yang valid: AnalyticDB for MySQL dan AnalyticDB for PostgreSQL.	MySQL
      AnalyticDB Instance ID	ID instans AnalyticDB for MySQL atau AnalyticDB for PostgreSQL tujuan.	am-bp139yqk8u1ik****
      Database Name	Nama database tujuan di instans AnalyticDB.	adb_demo
      Table Name	Nama tabel tujuan di database AnalyticDB untuk menyimpan pesan.	user
      Database Username	Username untuk menghubungkan ke instans AnalyticDB guna mengimpor data.	adbmysql
      Database Password	Password untuk menghubungkan ke instans AnalyticDB guna mengimpor data. Anda mengatur password saat membuat instans. Jika lupa password, Anda dapat mengatur ulang. AnalyticDB for MySQL: Untuk informasi cara mengatur ulang password, lihat Ubah password akun. AnalyticDB for PostgreSQL: Login ke Konsol AnalyticDB for PostgreSQL, klik nama instans, lalu klik Account Management di panel navigasi sebelah kiri. Temukan akun yang ingin Anda atur ulang password-nya, lalu klik Reset Password di kolom Actions.	********

      Catatan

      Message Queue for Apache Kafka meneruskan username dan password sebagai variabel lingkungan ke fungsi Function Compute saat membuat tugas. Setelah tugas dibuat, Message Queue for Apache Kafka tidak menyimpan informasi ini.

      Setelah konektor dibuat, Anda dapat melihatnya di halaman Connectors.

6. Pada halaman Connectors, temukan konektor yang telah Anda buat dan klik Deploy di kolom Actions.

Konfigurasikan layanan Function Compute

Setelah berhasil membuat dan mendeploy Konektor Sink AnalyticDB di konsol Message Queue for Apache Kafka, Function Compute secara otomatis membuat layanan dan fungsi untuk konektor tersebut. Layanan diberi nama dalam format kafka-service-<nama_konektor>-<string_acak>, dan fungsi diberi nama dalam format fc-adb-<string_acak>.

1. Pada halaman Connectors, temukan konektor target dan klik Configure Function di kolom Actions.

   Tindakan ini akan mengarahkan Anda ke konsol Function Compute.

2. Di Konsol Function Compute, temukan layanan fungsi yang dibuat secara otomatis dan konfigurasikan informasi VPC dan vSwitch-nya. Untuk langkah-langkah spesifik, lihat Perbarui layanan.

Konfigurasikan database AnalyticDB tujuan

Setelah mengonfigurasi layanan Function Compute, Anda harus menambahkan Blok CIDR vSwitch-nya ke daftar putih instans AnalyticDB Anda. Anda dapat menemukan Blok CIDR ini di halaman vSwitch di Konsol VPC, pada baris yang sesuai dengan VPC dan vSwitch layanan Function Compute.

• AnalyticDB for MySQL: Login ke AnalyticDB for MySQL console dan konfigurasikan daftar putih. Untuk informasi selengkapnya, lihat Konfigurasikan daftar putih alamat IP.

• AnalyticDB for PostgreSQL: Login ke Konsol AnalyticDB for PostgreSQL dan konfigurasikan daftar putih. Untuk informasi selengkapnya, lihat Konfigurasikan daftar putih.

Kirim pesan uji

1. Pada halaman Connectors, temukan konektor target dan klik Test di kolom Actions.

2. Di panel Send Message, kirim pesan uji.

   • Atur Sending Method ke Console.

     1. Di kotak teks Message Key, masukkan kunci pesan. Contohnya, demo.

     2. Di kotak teks Message Content, masukkan konten pesan uji. Contohnya, {"key": "test"}.

     3. Atur Send to Specified Partition untuk menentukan apakah akan mengirim pesan ke partisi tertentu.

        • Klik Yes dan masukkan ID partisi di kotak teks Partition ID. Contohnya, 0. Untuk mengkueri ID partisi, lihat Lihat status partisi.

        • Klik No untuk tidak menentukan partisi.

   • Atur Sending Method ke Docker. Jalankan perintah Docker di bagian Run the Docker container to produce a sample message untuk mengirim pesan.

   • Atur Sending Method ke SDK. Pilih SDK untuk bahasa atau framework yang diperlukan dan jenis koneksi untuk mengirim pesan.

Verifikasi hasil

Setelah mengirim pesan ke topik sumber di Message Queue for Apache Kafka, login ke AnalyticDB for MySQL console atau Konsol AnalyticDB for PostgreSQL, hubungkan ke database, lalu buka SQL Window di Data Management Service (DMS). Kueri tabel di instans yang sesuai untuk memeriksa apakah data telah diekspor dengan benar.