ApsaraMQ for Kafka:Buat konektor sink MaxCompute

Langkah 1: Buat tabel MaxCompute

Buat tabel tujuan di client MaxCompute. Untuk detailnya, lihat Buat tabel.

Dengan partisi — tambahkan kolom kunci partisi bernama time dengan tipe STRING:

CREATE TABLE IF NOT EXISTS kafka_to_maxcompute(topic STRING, valueName STRING, valueAge BIGINT) PARTITIONED by (time STRING);

Tanpa partisi:

CREATE TABLE IF NOT EXISTS kafka_to_maxcompute(topic STRING, valueName STRING, valueAge BIGINT);

Tip: Partisi mengorganisasi data ke dalam segmen berbasis waktu, yang meningkatkan performa kueri dan mengurangi biaya pemindaian untuk tabel besar. Aktifkan partisi jika Anda mengharapkan volume data tinggi atau perlu melakukan kueri berdasarkan rentang waktu. Lewati partisi untuk tabel kecil atau ketika penyaringan berbasis waktu tidak diperlukan.

Setelah pernyataan berhasil dijalankan, verifikasi tabel tersebut di halaman Tables di Konsol MaxCompute.

Langkah 2: Buat dan mulai konektor

1. Masuk ke ApsaraMQ for Kafka console. Pada bagian Resource Distribution di halaman Overview, pilih wilayah tempat instans Anda berada.

2. Di panel navigasi sebelah kiri, pilih Connector Ecosystem Integration > Tasks.

3. Di halaman Tasks, klik Create Task.

4. Di halaman Create Task, masukkan Task Name dan Description, lalu konfigurasikan bagian-bagian berikut.

{
  "data": {
    "topic": "t_test",
    "partition": 2,
    "offset": 1,
    "timestamp": 1717048990499,
    "headers": {
      "headers": [],
      "isReadOnly": false
    },
    "key": "MaxCompute-K1",
    "value": "MaxCompute-V1"
  },
  "id": "9b05fc19-9838-4990-bb49-ddb942307d3f-2-1",
  "source": "acs:alikafka",
  "specversion": "1.0",
  "type": "alikafka:Topic:Message",
  "datacontenttype": "application/json; charset=utf-8",
  "time": "2024-05-30T06:03:10.499Z",
  "aliyunaccountid": "1413397765616316"
}

Langkah 3: Verifikasi pengiriman data

Kirim pesan uji dan pastikan pesan tersebut tiba di tabel MaxCompute.

1. Di halaman Tasks, temukan konektor dan klik nama topik sumber di kolom Event Source.

2. Di halaman Detail Topik, klik Send Message.

3. Di panel Start to Send and Consume Message, masukkan pesan uji dan klik OK.

   

4. Di Konsol MaxCompute, kueri partisi untuk memastikan data telah tiba:

      show PARTITIONS kafka_to_maxcompute;

   

5. Kueri data di partisi target. Ganti nilai time dengan nilai partisi aktual dari langkah sebelumnya. Jika kueri mengembalikan data pesan uji, konektor berfungsi dengan benar.

      SELECT * FROM kafka_to_maxcompute WHERE time="2024-05-31.16:37.suffix";