ApsaraMQ for Kafka:Buat konektor sink OSS

Prasyarat

Catatan penggunaan

• Konektor mengekspor data dari topik sumber di instans ApsaraMQ for Kafka ke Function Compute, lalu dari Function Compute ke Object Storage Service (OSS). Instans ApsaraMQ for Kafka, layanan Function Compute, dan bucket OSS harus berada di wilayah yang sama. Untuk informasi selengkapnya mengenai batasan konektor, lihat Batasan.

• Fitur ini didasarkan pada layanan Function Compute. Function Compute menyediakan kuota gratis. Jika Anda melebihi kuota gratis tersebut, Anda akan dikenai biaya atas sumber daya tambahan yang digunakan. Untuk informasi selengkapnya mengenai penagihan, lihat Ikhtisar penagihan.

• Function Compute mendukung kueri log untuk pemanggilan fungsi. Untuk informasi selengkapnya, lihat Konfigurasikan log.

• Saat mengekspor pesan, ApsaraMQ for Kafka melakukan serialisasi dengan menggunakan encoding string UTF-8. Format data biner tidak didukung.

Buat dan deploy konektor sink OSS

1. Masuk ke Konsol ApsaraMQ for Kafka.

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 akan dibuat dari daftar drop-down Select Instance, lalu klik Create Connector.

5. Pada wizard Create Connector, lakukan langkah-langkah berikut:

   1. Pada langkah Configure Basic Information, konfigurasikan parameter dan klik Next. Tabel berikut menjelaskan parameter tersebut.

      Penting

      ApsaraMQ for Kafka secara otomatis memilih opsi Authorize to Create Service Linked Role.

      • Jika peran terkait layanan belum dibuat, ApsaraMQ for Kafka akan membuatnya secara otomatis untuk Anda. Peran ini memungkinkan ApsaraMQ for Kafka mengekspor data ke OSS.

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

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

      Parameter	Deskripsi	Contoh
      Name	Nama konektor. Nama harus memenuhi persyaratan berikut: Panjang nama maksimal 48 karakter dan hanya boleh terdiri dari angka, huruf kecil, dan tanda hubung (-). Nama tidak boleh dimulai dengan tanda hubung (-). Nama harus unik dalam satu instans ApsaraMQ for 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 secara otomatis untuk Anda.	kafka-oss-sink
      Instance	Secara default, nama dan ID instans ditampilkan.	demo alikafka_post-cn-st21p8vj****

   2. Pada langkah Configure Source Service, pilih Message Queue for Apache Kafka sebagai layanan sumber, konfigurasikan parameter, lalu klik Next. Tabel berikut menjelaskan parameter tersebut.

      Parameter	Deskripsi	Contoh
      Data Source Topic	Topik tempat data diekspor.	oss-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	Offset tempat konsumsi dimulai. Nilai yang valid: Earliest Offset: Konsumsi dimulai dari pesan pertama yang tersedia di partisi. Latest Offset: Konsumsi dimulai dari pesan berikutnya yang diproduksi setelah konektor dijalankan.	Earliest Offset
      VPC ID	VPC tempat tugas sinkronisasi data dijalankan. Klik Configure Runtime Environment untuk menampilkan parameter ini. Secara default, parameter ini diatur ke VPC instans ApsaraMQ for Kafka dan tidak perlu dikonfigurasi.	vpc-bp1xpdnd3l***
      vSwitch ID	vSwitch untuk tugas sinkronisasi data. Klik Configure Runtime Environment untuk menampilkan parameter ini. vSwitch harus berada di VPC yang sama dengan instans ApsaraMQ for Kafka. Secara default, ini adalah vSwitch yang ditentukan saat instans ApsaraMQ for Kafka dideploy.	vsw-bp1d2jgg81***
      Failure Handling Policy	Menentukan kebijakan penanganan kegagalan pengiriman pesan pada partisi topik. Klik Configure Runtime Environment untuk menampilkan parameter ini. Nilai yang valid: Continue Subscription: Melanjutkan berlangganan pada partisi topik tempat terjadi kesalahan dan mencetak log kesalahan. Stop Subscription: Menghentikan berlangganan pada partisi topik tempat terjadi kesalahan dan mencetak log kesalahan. Catatan Untuk informasi cara melihat log, lihat Operasi terkait konektor. Untuk informasi cara menemukan solusi berdasarkan kode kesalahan, lihat Kode kesalahan.	Continue Subscription
      Resource Creation Method	Metode pembuatan topik dan kelompok konsumen yang diperlukan oleh konektor. Klik Configure Runtime Environment untuk menampilkan parameter ini. Auto Manual	Auto
      Connector Consumer Group	Kelompok konsumen yang digunakan oleh konektor. Klik Configure Runtime Environment untuk menampilkan parameter ini. Disarankan agar nama kelompok konsumen ini diawali dengan connect-cluster.	connect-cluster-kafka-oss-sink
      Task Offset Topic	Topik untuk menyimpan offset konsumen. Klik Configure Runtime Environment untuk menampilkan parameter ini. Topik: Disarankan agar nama topik diawali dengan connect-offset. Jumlah partisi: Jumlah partisi harus lebih dari 1. Mesin penyimpanan: Mesin penyimpanan topik harus Local storage. Catatan Hanya instans Edisi Profesional yang mendukung pemilihan Local storage saat membuat topik. Instans Edisi Standar tidak mendukung fitur ini. cleanup.policy: Kebijakan pembersihan log untuk topik harus compact.	connect-offset-kafka-oss-sink
      Task Configuration Topic	Topik untuk menyimpan konfigurasi tugas. Klik Configure Runtime Environment untuk menampilkan parameter ini. Topik: Disarankan agar nama topik diawali dengan connect-config. Jumlah partisi: Jumlah partisi harus 1. Mesin penyimpanan: Mesin penyimpanan topik harus Local storage. Catatan Hanya instans Edisi Profesional yang mendukung pemilihan Local storage saat membuat topik. Instans Edisi Standar tidak mendukung fitur ini. cleanup.policy: Kebijakan pembersihan log untuk topik harus compact.	connect-config-kafka-oss-sink
      Task Status Topic	Topik untuk menyimpan status tugas. Klik Configure Runtime Environment untuk menampilkan parameter ini. Topik: Disarankan agar nama topik diawali dengan connect-status. Jumlah partisi: Disarankan jumlah partisi diatur ke 6. Mesin penyimpanan: Mesin penyimpanan topik harus Local storage. Catatan Hanya instans Edisi Profesional yang mendukung pemilihan Local storage saat membuat topik. Instans Edisi Standar tidak mendukung fitur ini. cleanup.policy: Kebijakan pembersihan log untuk topik harus compact.	connect-status-kafka-oss-sink
      Dead-letter Queue Topic	Topik untuk menyimpan data pengecualian dari framework Kafka Connect. Klik Configure Runtime Environment untuk menampilkan parameter ini. Topik ini dapat sama dengan Error Data Topic untuk menghemat sumber daya topik. Topik: Disarankan agar nama topik diawali dengan connect-error. Jumlah partisi: Disarankan jumlah partisi diatur ke 6. Mesin penyimpanan: Mesin penyimpanan topik dapat berupa Local storage atau Cloud storage. Catatan Hanya instans Edisi Profesional yang mendukung pemilihan Local storage saat membuat topik. Instans Edisi Standar tidak mendukung fitur ini.	connect-error-kafka-oss-sink
      Error Data Topic	Topik 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 agar nama topik diawali dengan connect-error. Jumlah partisi: Disarankan jumlah partisi diatur ke 6. Mesin penyimpanan: Mesin penyimpanan topik dapat berupa Local storage atau Cloud storage. Catatan Hanya instans Edisi Profesional yang mendukung pemilihan Local storage saat membuat topik. Instans Edisi Standar tidak mendukung fitur ini.	connect-error-kafka-oss-sink

   3. Pada langkah Configure Destination Service, pilih Object Storage Service sebagai layanan tujuan, atur parameter, lalu klik Create. Tabel berikut menjelaskan parameter tersebut.

      Parameter	Deskripsi	Contoh
      Bucket Name	Nama bucket OSS.	bucket_test
      AccessKey ID	ID AccessKey Akun Alibaba Cloud Anda.	yourAccessKeyID
      AccessKey Secret	Rahasia AccessKey Akun Alibaba Cloud Anda.	yourAccessKeySecret

      Pastikan Akun Alibaba Cloud Anda diberikan izin berikut sesuai prinsip hak istimewa minimal:

      {
          "Version": "1",
          "Statement": [
              {
                  "Action": [
                      "oss:GetObject",
                      "oss:PutObject"
                  ],
                  "Resource": "*",
                  "Effect": "Allow"
              }
          ]
      }

      Catatan

      ApsaraMQ for Kafka meneruskan ID AccessKey dan Rahasia AccessKey ke OSS sebagai variabel lingkungan saat membuat tugas. Setelah pembuatan, ApsaraMQ for Kafka tidak menyimpan kunci-kunci tersebut.

      Setelah konektor dibuat, Anda dapat melihatnya di halaman Connectors.

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

Kirim pesan

Anda dapat mengirim pesan ke topik sumber di ApsaraMQ for Kafka untuk memverifikasi bahwa data telah diekspor ke OSS.

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

2. Di panel Send Message, kirim pesan uji coba.

   • 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 coba. Contohnya, {"key": "test"}.

     3. Atur Send to Specified Partition untuk menentukan apakah pesan dikirim 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 dibutuhkan dan jenis koneksi untuk mengirim pesan.

Verifikasi hasil

Setelah Anda mengirim pesan ke topik sumber di ApsaraMQ for Kafka, periksa file manager OSS untuk memverifikasi bahwa data telah diekspor. Untuk informasi selengkapnya, lihat Ikhtisar file.

Jika objek baru dihasilkan di bucket OSS, data telah disinkronkan ke OSS.

[
    {
        "key":"123",
        "offset":4,
        "overflowFlag":true,
        "partition":0,
        "timestamp":1603779578478,
        "topic":"Test",
        "value":"1",
        "valueSize":272687
    }
]

Operasi terkait

Anda dapat mengonfigurasi sumber daya Function Compute yang diperlukan oleh konektor sink OSS sesuai kebutuhan.