ApsaraMQ for Kafka:Buat Konektor Sink Tablestore

Catatan

• Data hanya dapat diekspor dari topik sumber dalam instans ApsaraMQ for Kafka ke Tablestore dalam wilayah yang sama. Untuk informasi tentang batasan konektor, lihat Batasan penggunaan.

• Saat membuat konektor, ApsaraMQ for Kafka secara otomatis membuat peran terkait layanan.

  • Jika peran terkait layanan belum tersedia, ApsaraMQ for Kafka akan membuatnya secara otomatis sehingga memungkinkan ApsaraMQ for Kafka mengekspor data ke Tablestore.

  • Jika peran terkait layanan sudah ada, ApsaraMQ for Kafka tidak membuat yang baru.

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

Prosedur

Topik ini menjelaskan cara menggunakan konektor sink Tablestore untuk mengekspor data dari topik sumber dalam instans ApsaraMQ for Kafka ke Tablestore.

1. Opsi: Buat topik dan group yang diperlukan untuk konektor sink Tablestore.

   Jika Anda tidak perlu menyesuaikan topik dan group, langkah ini dapat dilewati dengan memilih opsi Auto pada langkah berikutnya.

   Penting

   Beberapa topik yang diperlukan oleh konektor sink Tablestore harus menggunakan mesin penyimpanan Local. Untuk instans ApsaraMQ for Kafka versi utama 0.10.2, Anda tidak dapat membuat topik yang menggunakan mesin penyimpanan Local secara manual—topik tersebut hanya dapat dibuat secara otomatis.

   1. Buat topik untuk konektor sink Tablestore

   2. Buat group untuk konektor sink Tablestore

2. Buat dan deploy konektor sink Tablestore

3. Verifikasi hasilnya

   1. Kirim pesan uji

   2. Lihat data dalam tabel

Buat topik untuk konektor sink Tablestore

Di konsol ApsaraMQ for Kafka, buat secara manual lima topik yang diperlukan untuk konektor sink Tablestore: topik offset tugas, topik konfigurasi tugas, topik status tugas, topik dead-letter queue, dan topik data error. Jumlah partisi dan mesin penyimpanan yang diperlukan berbeda untuk setiap topik. Untuk informasi selengkapnya, lihat Daftar parameter layanan sumber.

1. Masuk ke ApsaraMQ for Kafka console.

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

   Penting

   Topik harus dibuat dalam wilayah yang sama dengan aplikasi Anda, yaitu wilayah tempat Instance ECS dideploy. Topik tidak dapat digunakan lintas wilayah. Misalnya, jika topik dibuat di wilayah China (Beijing), produsen dan konsumen pesan juga harus berjalan pada Instance ECS di wilayah China (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 replika terdistribusi tiga arah. Jika Instance Edition instans adalah Standard (High Write), mesin penyimpanan hanya dapat diatur ke Cloud Storage. Local Storage: Menggunakan algoritma replikasi in-sync replica (ISR) Kafka asli dan mekanisme replika terdistribusi tiga arah.	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 dalam kluster mengalami gangguan, 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 dalam kluster mengalami gangguan, 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 disimpan. Ini terutama berlaku untuk skenario seperti memulihkan status setelah sistem mengalami gangguan atau memuat ulang cache setelah sistem direstart. Misalnya, saat menggunakan Kafka Connect atau Confluent Schema Registry, Anda harus menggunakan Kafka Compact Topic untuk menyimpan status sistem atau informasi konfigurasi. 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 untuk konektor sink Tablestore

Anda dapat membuat group secara manual untuk tugas sinkronisasi data konektor sink Tablestore di konsol ApsaraMQ for Kafka. Nama group harus mengikuti format connect-Nama Tugas. Untuk informasi selengkapnya, lihat Parameter layanan sumber.

1. Masuk 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 group di kotak teks Group ID, masukkan deskripsi singkat di kotak teks Description, tambahkan tag ke group, lalu klik OK.

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

Buat dan deploy konektor sink Tablestore

Buat dan deploy konektor sink Tablestore untuk menyinkronkan data dari instans ApsaraMQ for Kafka ke tabel Tablestore.

1. Masuk 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 berikut lalu klik Next.

      Parameter	Deskripsi	Contoh
      Name	Nama konektor. Nama harus memenuhi persyaratan berikut: Nama dapat terdiri dari maksimal 48 karakter dan hanya boleh berisi angka, huruf kecil, dan tanda hubung (-). Nama tidak boleh dimulai dengan tanda hubung (-). Nama harus unik dalam satu instans ApsaraMQ for Kafka. Tugas sinkronisasi data konektor menggunakan consumer group bernama connect-task-name. Jika Anda tidak membuat consumer group ini secara manual, sistem akan membuatnya secara otomatis.	kafka-ts-sink
      Instance	Secara default, nama instans dan ID instans ditampilkan.	demo alikafka_post-cn-st21p8vj****

   2. Pada tab Configure Source Service, atur Data Source ke ApsaraMQ for Kafka, konfigurasikan parameter berikut, lalu klik Next.

      Catatan

      Jika Anda telah membuat topik dan group yang diperlukan, pilih Manual untuk pembuatan resource dan masukkan informasinya. Jika tidak, pilih Auto.

      Tabel 1. Parameter untuk layanan sumber

      Parameter	Deskripsi	Contoh
      Data Source Topic	Topik sumber untuk sinkronisasi data.	ts-test-input
      Consumer Thread Concurrency	Jumlah thread konsumen konkuren untuk topik sumber data. Nilai default adalah 6. Nilai yang valid: 1 2 3 6 12	6
      Consumer Offset	Posisi awal konsumsi pesan. Nilai yang valid: Earliest Offset: Mengonsumsi pesan dari offset paling awal. Latest Offset: Mengonsumsi pesan dari offset terbaru.	Earliest Offset
      VPC ID	ID virtual private cloud (VPC) tempat tugas sinkronisasi data berjalan. Parameter ini ditampilkan setelah Anda mengklik Configure Runtime Environment. Parameter ini secara default menggunakan VPC dari instans ApsaraMQ for Kafka dan tidak perlu dikonfigurasi.	vpc-bp1xpdnd3l***
      vSwitch ID	ID vSwitch tempat tugas sinkronisasi data berjalan. Parameter ini ditampilkan setelah Anda mengklik Configure Runtime Environment. vSwitch harus berada dalam VPC yang sama dengan instans ApsaraMQ for Kafka. Parameter ini secara default menggunakan vSwitch yang ditentukan saat instans ApsaraMQ for Kafka dideploy.	vsw-bp1d2jgg81***
      Failure Handling Policy	Kebijakan penanganan kegagalan pengiriman pesan dalam partisi topik. Parameter ini ditampilkan setelah Anda mengklik Configure Runtime Environment. Nilai yang valid: Continue Subscription: Melanjutkan langganan ke partisi topik tempat terjadi kesalahan dan mencetak log kesalahan. Stop Subscription: Menghentikan langganan ke partisi topik tempat terjadi kesalahan dan mencetak log kesalahan. Catatan Untuk informasi selengkapnya tentang cara melihat log, lihat Kelola konektor. Untuk informasi selengkapnya tentang cara menemukan solusi berdasarkan kode kesalahan, lihat Kode kesalahan.	Continue Subscription
      Resource Creation Method	Metode pembuatan topik dan group yang diperlukan. Parameter ini ditampilkan setelah Anda mengklik Configure Runtime Environment. Auto Manual	Auto
      Connector Consumer Group	Group konsumen yang digunakan oleh tugas sinkronisasi data. Parameter ini ditampilkan setelah Anda mengklik Configure Runtime Environment. Nama group harus berformat connect-task-name.	connect-cluster-kafka-ots-sink
      Task Offset Topic	Topik yang menyimpan offset konsumen. Parameter ini ditampilkan setelah Anda mengklik Configure Runtime Environment. Topik: Nama topik harus dimulai dengan connect-offset. Partisi: Jumlah partisi untuk topik harus lebih dari 1. Mesin penyimpanan: Mesin penyimpanan untuk topik harus Local Storage. cleanup.policy: Kebijakan pembersihan log untuk topik harus compact.	connect-offset-kafka-ots-sink
      Task Configuration Topic	Topik yang menyimpan konfigurasi tugas. Parameter ini ditampilkan setelah Anda mengklik Configure Runtime Environment. Topik: Nama topik harus dimulai dengan connect-config. Partisi: Jumlah partisi untuk topik harus 1. Mesin penyimpanan: Mesin penyimpanan untuk topik harus Local Storage. cleanup.policy: Kebijakan pembersihan log untuk topik harus compact.	connect-config-kafka-ots-sink
      Task Status Topic	Topik yang menyimpan status tugas. Parameter ini ditampilkan setelah Anda mengklik Configure Runtime Environment. Topik: Nama topik harus dimulai dengan connect-status. Partisi: Jumlah partisi yang disarankan adalah 6. Mesin penyimpanan: Mesin penyimpanan untuk topik harus Local Storage. cleanup.policy: Kebijakan pembersihan log untuk topik harus compact.	connect-status-kafka-ots-sink
      Dead-letter Queue Topic	Topik yang digunakan untuk menyimpan data error dari framework Kafka Connect. Parameter ini ditampilkan setelah Anda mengklik Configure Runtime Environment. Untuk menghemat resource topik, Anda dapat menggunakan topik yang sama untuk dead-letter queue dan Error Data Topic. Topik: Nama topik harus dimulai dengan connect-error. Partisi: Jumlah partisi yang disarankan adalah 6. Mesin penyimpanan: Mesin penyimpanan untuk topik dapat berupa Local Storage atau Cloud Storage.	connect-error-kafka-ots-sink
      Error Data Topic	Topik yang digunakan untuk menyimpan data error dari sink. Parameter ini ditampilkan setelah Anda mengklik Configure Runtime Environment. Untuk menghemat resource topik, Anda dapat menggunakan topik yang sama untuk Dead-letter Queue Topic dan topik data error. Topik: Nama topik harus dimulai dengan connect-error. Partisi: Jumlah partisi yang disarankan adalah 6. Mesin penyimpanan: Mesin penyimpanan untuk topik dapat berupa Local Storage atau Cloud Storage.	connect-error-kafka-ots-sink

   3. Pada tab Configure Destination Service, atur Destination Service ke Tablestore, konfigurasikan parameter berikut, lalu klik Create.

      Parameter	Deskripsi	Contoh
      Instance Name	Nama instans Tablestore.	k00eny67****
      Automatically Create Destination Table	Menentukan apakah akan membuat tabel secara otomatis di Tablestore. Yes: Tabel dibuat secara otomatis di Tablestore untuk menyimpan data yang disinkronkan berdasarkan nama tabel yang dikonfigurasi. No: Tabel yang sudah ada digunakan untuk menyimpan data yang disinkronkan.	Yes
      Destination Table Name	Nama tabel yang menyimpan data yang disinkronkan. Jika Anda mengatur Automatically Create Destination Table ke No, nama tabel harus sama dengan nama tabel yang sudah ada di instans Tablestore.	kafka_table
      Tablestore	Jenis tabel yang menyimpan data yang disinkronkan. Wide Column Model TimeSeries Model	Wide Column Model
      Message Key Format	Format input kunci pesan. Nilai yang valid adalah String dan JSON. Nilai default adalah JSON. Parameter ini hanya ditampilkan ketika Tablestore diatur ke Wide Column Model. String: Kunci pesan diurai langsung sebagai string. JSON: Kunci pesan harus dalam format JSON.	String
      Message Value Format	Format input nilai pesan. Nilai yang valid adalah String dan JSON. Nilai default adalah JSON. Parameter ini hanya ditampilkan ketika Tablestore diatur ke Wide Column Model. String: Nilai pesan diurai langsung sebagai string. JSON: Nilai pesan harus dalam format JSON.	String
      JSON Message Field Conversion	Metode pemrosesan field dalam pesan JSON. Parameter ini ditampilkan jika Anda mengatur Message Key Format atau Message Value Format ke JSON. Nilai yang valid: Write All as String: Mengonversi semua field ke tipe String di Tablestore. Automatically Identify Field Types: Mengonversi field String dan Boolean dalam isi pesan JSON ke tipe String dan Boolean yang sesuai di Tablestore. Tipe data Integer dan Float dalam isi pesan JSON dikonversi ke tipe Double di Tablestore.	Write All as String
      Primary Key Mode	Menentukan mode kunci primer. Anda dapat mengekstraksi kunci primer tabel dari berbagai bagian catatan pesan ApsaraMQ for Kafka, termasuk koordinat (topik, partisi, dan offset), kunci, dan nilai. Parameter ini hanya ditampilkan ketika Tablestore diatur ke Wide Column Model. Nilai default adalah kafka. kafka: Menggunakan <connect_topic>_<connect_partition> dan <connect_offset> sebagai kunci primer tabel data. record_key: Menggunakan field dalam kunci catatan sebagai kunci primer tabel data. record_value: Menggunakan field dalam nilai catatan sebagai kunci primer tabel data.	kafka
      Primary Key Column Names	Nama kolom kunci primer tabel data dan tipe data yang sesuai. Tipe data String dan Integer didukung. Artinya, field dalam kunci atau nilai catatan yang sesuai dengan nama kolom yang dikonfigurasi digunakan sebagai kunci primer untuk tabel data. Parameter ini ditampilkan jika Anda mengatur Message Key Format ke JSON dan Primary Key Mode ke record_key, atau jika Anda mengatur Message Value Format ke JSON dan Primary Key Mode ke record_value. Klik Create untuk menambahkan nama kolom. Anda dapat mengonfigurasi maksimal empat nama kolom.	None
      Write Mode	Menentukan mode penulisan. Nilai yang valid adalah put dan update. Nilai default adalah put. Parameter ini hanya ditampilkan ketika Tablestore diatur ke Wide Column Model. put: Menimpa data yang sudah ada. update: Memperbarui data yang sudah ada.	put
      Delete Mode	Jika catatan pesan ApsaraMQ for Kafka berisi nilai null, Anda dapat memilih apakah akan menghapus baris atau kolom atribut. Parameter ini ditampilkan ketika Primary Key Mode diatur ke record_key. Nilai yang valid: none: Nilai default. Tidak ada penghapusan yang diizinkan. row: Mengizinkan penghapusan baris. column: Mengizinkan penghapusan kolom atribut. row_and_column: Mengizinkan penghapusan baris dan kolom atribut. Perilaku penghapusan bergantung pada mode penulisan: Jika Write Mode adalah put, semua mode penghapusan menghasilkan penulisan timpa ke tabel data Tablestore, bahkan ketika nilai berisi field null. Jika Write Mode adalah update dan mode penghapusan adalah none atau row, catatan yang semua field nilainya null dianggap sebagai data rusak. Jika hanya beberapa field nilai yang null, konektor mengabaikan field null tersebut dan menulis field non-null ke tabel data Tablestore. Jika mode penghapusan adalah column atau row_and_column, konektor menghapus baris dan kolom atribut untuk field null lalu menulis data ke tabel data Tablestore.	None
      Metric Name Field	Field ini dipetakan ke field nama metrik (_m_name) dalam Model TimeSeries Tablestore. Nama metrik menentukan nama besaran fisik atau metrik pemantauan untuk data deret waktu, seperti suhu atau kecepatan. Field ini tidak boleh kosong. Parameter ini ditampilkan ketika Anda memilih Model TimeSeries untuk Tablestore.	measurement
      Data Source Field	Petakan field ini ke field sumber data (_data_source) dalam Model TimeSeries Tablestore. Field ini berfungsi sebagai pengenal sumber data deret waktu, seperti nama mesin atau ID perangkat, dan dapat dikosongkan. Parameter ini ditampilkan ketika Model TimeSeries dipilih untuk Tablestore.	source
      Tag Field	Gunakan satu atau beberapa field sebagai field tag (_tags) untuk Model TimeSeries Tablestore. Setiap tag adalah pasangan kunci-nilai string. Kuncinya adalah nama field yang dikonfigurasi, dan nilainya adalah isi field tersebut. Tag merupakan bagian dari metadata deret waktu. Deret waktu diidentifikasi secara unik oleh kombinasi nama metrik, sumber data, dan tag. Tag dapat dikosongkan. Parameter ini ditampilkan ketika Tablestore dipilih sebagai model deret waktu.	tag1, tag2
      Timestamp Field	Petakan field ini ke field timestamp (_time) dalam Model TimeSeries Tablestore. Field ini merepresentasikan titik waktu untuk baris data deret waktu ini, seperti waktu saat besaran fisik dihasilkan. Saat data ditulis ke Tablestore, field timestamp dikonversi ke mikrodetik untuk penulisan dan penyimpanan. Parameter ini ditampilkan ketika Anda memilih Model TimeSeries untuk Tablestore.	time
      Timestamp Unit	Konfigurasikan parameter ini berdasarkan field timestamp aktual. Parameter ini hanya ditampilkan ketika Tablestore diatur ke TimeSeries Model. Nilai yang valid: SECONDS MILLISECONDS MICROSECONDS NANOSECONDS	MILLISECONDS
      Whether to Map All Non-primary Key Fields	Menentukan apakah akan memetakan semua field non-kunci primer sebagai field data. Field non-kunci primer adalah field yang belum dipetakan sebagai nama metrik, sumber data, tag, atau timestamp. Parameter ini hanya ditampilkan ketika Tablestore diatur ke TimeSeries Model. Nilai yang valid: Yes: Field dipetakan secara otomatis dan tipe datanya ditentukan. Tipe numerik semuanya dikonversi ke tipe Double. No: Anda harus menentukan field dan tipe yang akan dipetakan.	Yes
      Configure Mapping for All Non-primary Key Fields	Tipe data yang sesuai dengan nama field non-kunci primer tabel deret waktu. Tipe data Double, Integer, String, Binary, dan Boolean didukung. Parameter ini ditampilkan jika Anda mengatur Whether to Map All Non-primary Key Fields ke No.	String

      Setelah konektor dibuat, Anda dapat melihatnya di halaman Connectors.

6. Setelah konektor dibuat, temukan di halaman Connectors dan klik Deploy di kolom Actions.

7. Klik OK.

Kirim pesan uji

Setelah mendeploy konektor sink Tablestore, Anda dapat mengirim pesan ke topik sumber di ApsaraMQ for Kafka untuk memverifikasi bahwa data telah disinkronkan ke Tablestore.

1. Di 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 isi 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 mengetahui 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.

Lihat data tabel