Cara mengelola pemicu DTS di konsol Function Compute - Function Compute

Data Transmission Service (DTS) dapat diintegrasikan dengan EventBridge sebagai sumber event. Setelah DTS terintegrasi dengan Function Compute, pemicu DTS dapat digunakan untuk memicu eksekusi fungsi dalam Function Compute. Anda dapat menggunakan fungsi ini untuk memproses data basis data tambahan secara real-time yang diperoleh dari tugas pelacakan perubahan DTS. Topik ini menjelaskan cara membuat pemicu DTS di konsol Function Compute, mengonfigurasi parameter input, serta menulis dan menguji kode.

Ikhtisar

Setelah Anda mengirimkan permintaan untuk membuat pemicu di konsol Function Compute, Function Compute akan membuat sumber daya aliran event di sisi EventBridge berdasarkan konfigurasi pemicu.

Setelah pemicu dibuat, Anda dapat melihat informasi terkait di konsol Function Compute. Informasi tentang sumber daya yang dibuat secara otomatis juga dapat dilihat di konsol EventBridge. Ketika tugas pelacakan perubahan DTS menangkap data tambahan dari basis data, eksekusi fungsi dipicu dan satu atau lebih pesan event didorong ke fungsi dalam batch sesuai dengan konfigurasi batch.

Catatan Penggunaan

Tugas pelacakan perubahan DTS yang digunakan sebagai sumber pemicu harus berada di wilayah yang sama dengan fungsi di Function Compute.
Jika jumlah aliran event yang dibuat mencapai batas maksimum, pemicu DTS tidak dapat dibuat. Untuk informasi lebih lanjut tentang batasan aliran event, lihat Batasan.

Sebelum Memulai

EventBridge
- Aktifkan EventBridge dan berikan izin kepada pengguna RAM
Function Compute
- Buat layanan
- Buat fungsi
DTS
- Buat tugas pelacakan perubahan
- Buat grup konsumen

Langkah 1: Buat Pemicu DTS

Masuk ke Konsol Function Compute. Di panel navigasi sebelah kiri, klik Services & Functions.
Di bilah navigasi atas, pilih wilayah. Di halaman Services, klik layanan yang diinginkan.
Di halaman Functions, klik fungsi yang ingin Anda kelola.
Di halaman detail fungsi, klik tab Triggers, pilih versi atau alias dari daftar drop-down Version or Alias, lalu klik Create Trigger.

Di panel Buat Pemicu, tentukan parameter terkait. Setelah menentukan parameter, klik OK.

Tabel berikut menjelaskan parameter dasar.

Parameter	Deskripsi	Contoh
Trigger Type	Jenis pemicu. Untuk informasi lebih lanjut tentang jenis pemicu yang didukung, lihat Ikhtisar Pemicu.	DTS
Name	Nama pemicu.	dts-trigger
Version or Alias	Nilai default adalah LATEST. Jika Anda ingin membuat pemicu untuk versi atau alias lainnya, pilih versi atau alias di pojok kanan atas halaman detail fungsi. Untuk informasi lebih lanjut tentang versi dan alias layanan, lihat Kelola versi dan Kelola alias.	LATEST
Change Tracking Task	Nama tugas pelacakan perubahan.	dtsqntc2***
Consumer Group	Nama grup konsumen yang Anda buat untuk mengonsumsi data dari tugas pelacakan perubahan. Penting Pastikan bahwa grup konsumen hanya berjalan pada satu klien. Jika tidak, offset konsumen yang ditentukan mungkin menjadi tidak valid.	test
Account	Nama akun yang ditentukan saat grup konsumen dibuat.	test
Password	Kata sandi akun yang ditentukan saat grup konsumen dibuat.	******
Consumer Offset	Timestamp entri data pertama yang akan dikonsumsi. Entri data yang ditentukan oleh offset konsumen harus berada dalam rentang data tugas pelacakan perubahan. Catatan Offset konsumen hanya berlaku pada saat grup konsumen baru pertama kali dijalankan. Jika restart dilakukan di tugas selanjutnya, konsumsi akan dilanjutkan berdasarkan offset konsumen terakhir.	2022-06-21 00:00:00
Invocation Method	Pilih metode untuk memanggil fungsi. Nilai valid: Sync Invocation: Mode ini cocok untuk pemanggilan berurutan. Ketika sebuah event atau sekelompok event memicu fungsi, Function Compute menjalankan fungsi dan menunggu respons sebelum memproses event atau sekelompok event berikutnya. Batas atas payload untuk permintaan pemanggilan sinkron adalah 32 MB. Untuk informasi lebih lanjut, lihat Pemanggilan Sinkron. Async Invocation: Mode ini memungkinkan Anda untuk dengan cepat mengonsumsi event. Ketika sebuah event tunggal atau sekelompok event memicu fungsi, Function Compute segera mengembalikan respons dan melanjutkan memproses event atau sekelompok event berikutnya. Selama proses ini, fungsi dijalankan dalam mode asinkron. Batas atas payload untuk permintaan pemanggilan asinkron adalah 128 KB. Untuk informasi lebih lanjut, lihat Ikhtisar.	Pemanggilan Sinkron
Trigger State	Tentukan apakah akan mengaktifkan pemicu setelah dibuat. Secara default, Enable Trigger dipilih dan pemicu diaktifkan setelah dibuat.	N/A

Untuk informasi lebih lanjut tentang konfigurasi lanjutan seperti pengiriman pesan, percobaan ulang, dan konfigurasi surat mati, lihat Fitur lanjutan pemicu.

Setelah pemicu dibuat, pemicu tersebut ditampilkan di tab Triggers. Untuk memodifikasi atau menghapus pemicu yang ada, lihat Kelola pemicu.

Langkah 2: Konfigurasikan parameter input fungsi

Sumber event DTS dilewatkan ke fungsi dalam bentuk event, yang bertindak sebagai parameter input. Anda dapat melewatkan event secara manual ke fungsi untuk memicu fungsi tersebut.

Di halaman detail fungsi, klik tab Code dan klik ikon . Dari daftar drop-down yang muncul, pilih Configure Test Parameters.

Di panel Configure Test Parameters, klik tab Create New Test Event atau Modify Existing Test Event, dan tentukan Event Name serta isi event. Setelah menentukan parameter, klik OK.

Contoh kode event:

[
  {
    "data": {
      "id": 321****,
      "topicPartition": {
        "hash": 0,
        "partition": 0,
        "topic": "cn_hangzhou_rm_1234****_test_version2"
      },
      "offset": 3218099,
      "sourceTimestamp": 1654847757,
      "operationType": "UPDATE",
      "schema": {
        "recordFields": [
          {
            "fieldName": "id",
            "rawDataTypeNum": 8,
            "isPrimaryKey": true,
            "isUniqueKey": false,
            "fieldPosition": 0
          },
          {
            "fieldName": "topic",
            "rawDataTypeNum": 253,
            "isPrimaryKey": false,
            "isUniqueKey": false,
            "fieldPosition": 1
          }
        ],
        "nameIndex": {
          "id": {
            "fieldName": "id",
            "rawDataTypeNum": 8,
            "isPrimaryKey": true,
            "isUniqueKey": false,
            "fieldPosition": 0
          },
          "topic": {
            "fieldName": "topic",
            "rawDataTypeNum": 253,
            "isPrimaryKey": false,
            "isUniqueKey": false,
            "fieldPosition": 1
          }
        },
        "schemaId": "(hangzhou-test-db,hangzhou-test-db,message_info)",
        "databaseName": "hangzhou--test-db",
        "tableName": "message_info",
        "primaryIndexInfo": {
          "indexType": "PrimaryKey",
          "indexFields": [
            {
              "fieldName": "id",
              "rawDataTypeNum": 8,
              "isPrimaryKey": true,
              "isUniqueKey": false,
              "fieldPosition": 0
            }
          ],
          "cardinality": 0,
          "nullable": true,
          "isFirstUniqueIndex": false
        },
        "uniqueIndexInfo": [],
        "foreignIndexInfo": [],
        "normalIndexInfo": [],
        "databaseInfo": {
          "databaseType": "MySQL",
          "version": "5.7.35-log"
        },
        "totalRows": 0
      },
      "beforeImage": {
        "recordSchema": {
          "recordFields": [
            {
              "fieldName": "id",
              "rawDataTypeNum": 8,
              "isPrimaryKey": true,
              "isUniqueKey": false,
              "fieldPosition": 0
            },
            {
              "fieldName": "topic",
              "rawDataTypeNum": 253,
              "isPrimaryKey": false,
              "isUniqueKey": false,
              "fieldPosition": 1
            }
          ],
          "nameIndex": {
            "id": {
              "fieldName": "id",
              "rawDataTypeNum": 8,
              "isPrimaryKey": true,
              "isUniqueKey": false,
              "fieldPosition": 0
            },
            "topic": {
              "fieldName": "topic",
              "rawDataTypeNum": 253,
              "isPrimaryKey": false,
              "isUniqueKey": false,
              "fieldPosition": 1
            }
          },
          "schemaId": "(hangzhou-test-db,hangzhou-test-db,message_info)",
          "databaseName": "hangzhou-test-db",
          "tableName": "message_info",
          "primaryIndexInfo": {
            "indexType": "PrimaryKey",
            "indexFields": [
              {
                "fieldName": "id",
                "rawDataTypeNum": 8,
                "isPrimaryKey": true,
                "isUniqueKey": false,
                "fieldPosition": 0
              }
                ],
                "cardinality": 0,
                "nullable": true,
                "isFirstUniqueIndex": false
              },
                "uniqueIndexInfo": [],
                "foreignIndexInfo": [],
                "normalIndexInfo": [],
                "databaseInfo": {
                "databaseType": "MySQL",
                "version": "5.7.35-log"
              },
                "totalRows": 0
              },
                "values": [
                {
                "data": 115
              },
                {
                "data": {
                "hb": [
                104,
                101,
                108,
                108,
                111
                ],
                "offset": 0,
                "isReadOnly": false,
                "bigEndian": true,
                "nativeByteOrder": false,
                "mark": -1,
                "position": 0,
                "limit": 9,
                "capacity": 9,
                "address": 0
              },
                "charset": "utf8mb4"
              }
                ],
                "size": 45
              },
                "afterImage": {
                "recordSchema": {
                "recordFields": [
                {
                "fieldName": "id",
                "rawDataTypeNum": 8,
                "isPrimaryKey": true,
                "isUniqueKey": false,
                "fieldPosition": 0
              },
                {
                "fieldName": "topic",
                "rawDataTypeNum": 253,
                "isPrimaryKey": false,
                "isUniqueKey": false,
                "fieldPosition": 1
              }
                ],
                "nameIndex": {
                "id": {
                "fieldName": "id",
                "rawDataTypeNum": 8,
                "isPrimaryKey": true,
                "isUniqueKey": false,
                "fieldPosition": 0
              },
                "topic": {
                "fieldName": "topic",
                "rawDataTypeNum": 253,
                "isPrimaryKey": false,
                "isUniqueKey": false,
                "fieldPosition": 1
              }
              },
                "schemaId": "(hangzhou-test-db,hangzhou-test-db,message_info)",
                "databaseName": "hangzhou-test-db",
                "tableName": "message_info",
                "primaryIndexInfo": {
                "indexType": "PrimaryKey",
                "indexFields": [
                {
                "fieldName": "id",
                "rawDataTypeNum": 8,
                "isPrimaryKey": true,
                "isUniqueKey": false,
                "fieldPosition": 0
              }
                ],
                "cardinality": 0,
                "nullable": true,
                "isFirstUniqueIndex": false
              },
                "uniqueIndexInfo": [],
                "foreignIndexInfo": [],
                "normalIndexInfo": [],
                "databaseInfo": {
                "databaseType": "MySQL",
                "version": "5.7.35-log"
              },
                "totalRows": 0
              },
                "values": [
                {
                "data": 115
              },
                {
                "data": {
                "hb": [
                98,
                121,
                101
                ],
                "offset": 0,
                "isReadOnly": false,
                "bigEndian": true,
                "nativeByteOrder": false,
                "mark": -1,
                "position": 0,
                "limit": 11,
                "capacity": 11,
                "address": 0
              },
                "charset": "utf8mb4"
              }
                ],
                "size": 47
              }
              },
    "id": "12f701a43741d404fa9a7be89d9acae0-321****",
    "source": "DTSstreamDemo",
    "specversion": "1.0",
    "type": "dts:ConsumeMessage",
    "datacontenttype": "application/json; charset=utf-8",
    "time": "2022-06-10T07:55:57Z",
    "subject": "acs:dts:cn-hangzhou:12345****:kk123abc60g782/dtsabcdet1ro"
  }
]

Untuk informasi tentang parameter yang didefinisikan dalam spesifikasi CloudEvents, lihat Ikhtisar.

Tabel berikut menjelaskan parameter yang terkandung dalam bidang data.

Parameter	Tipe	Deskripsi
id	String	ID entri data DTS.
topicPartition	Array	Informasi partisi tentang topik tempat event didorong.
hash	String	Parameter penyimpanan bawah DTS.
partition	String	Partisi.
topic	String	Nama topik.
offset	Int	Offset entri data DTS.
sourceTimestamp	Int	Timestamp yang menunjukkan kapan entri data DTS dibuat.
operationType	String	Jenis operasi yang terlibat dalam entri data DTS.
schema	Array	Informasi skema tentang basis data.
recordFields	Array	Rincian bidang.
fieldName	String	Nama bidang.
rawDataTypeNum	Int	Nilai yang dipetakan dari tipe bidang. Nilai parameter ini sesuai dengan nilai bidang dataTypeNumber dalam data tambahan yang telah dideserialisasi dari instance pelacakan perubahan. Untuk informasi lebih lanjut, lihat Gunakan klien Kafka untuk mengonsumsi data terlacak.
isPrimaryKey	Boolean	Menunjukkan apakah bidang tersebut adalah bidang kunci utama.
isUniqueKey	Boolean	Menunjukkan apakah bidang tersebut memiliki kunci unik.
fieldPosition	String	Posisi bidang.
nameIndex	Array	Informasi pengindeksan bidang berdasarkan nama bidang.
schemaId	String	ID skema basis data.
databaseName	String	Nama basis data.
tableName	String	Nama tabel.
primaryIndexInfo	String	Indeks kunci utama.
indexType	String	Tipe indeks.
indexFields	Array	Bidang tempat indeks dibuat.
cardinality	String	Kardinalitas kunci utama.
nullable	Boolean	Menunjukkan apakah kunci utama dapat bernilai null.
isFirstUniqueIndex	Boolean	Menunjukkan apakah indeks tersebut adalah indeks unik pertama.
uniqueIndexInfo	String	Indeks unik.
foreignIndexInfo	String	Indeks untuk kunci asing.
normalIndexInfo	String	Indeks reguler.
databaseInfo	Array	Informasi tentang basis data.
databaseType	String	Mesin basis data.
version	String	Versi mesin basis data.
totalRows	Int	Jumlah total baris dalam tabel.
beforeImage	String	Gambar yang mencatat nilai bidang sebelum operasi dilakukan.
values	String	Nilai bidang yang dicatat.
size	Int	Ukuran bidang yang dicatat.
afterImage	String	Gambar yang mencatat nilai bidang setelah operasi dilakukan.

Langkah 3: Tulis dan uji fungsi

Setelah membuat pemicu, Anda dapat menulis kode fungsi dan menguji fungsi tersebut untuk memverifikasi kebenarannya. Dalam skenario nyata, ketika tugas pelacakan perubahan DTS menangkap data tambahan dari basis data, pemicu secara otomatis memicu eksekusi fungsi.

Di halaman detail fungsi, klik tab Code, edit kode fungsi di editor kode, lalu klik Deploy.

Berikut adalah contoh kode fungsi Node.js.

'use strict';
/*
Untuk mengaktifkan fitur inisialisasi
silakan implementasikan fungsi inisialisasi sebagai berikut:
exports.initializer = (context, callback) => {
  console.log('initializing');
  callback(null, '');
};
*/
exports.handler = (event, context, callback) => {
  console.log("event: %s", event);
  // Parsing parameter event dan memproses event.
  callback(null, 'hasil return');
}

Klik tab Code dan klik Test Function.
Setelah fungsi dieksekusi, hasilnya dapat dilihat di tab Kode.

Referensi

Selain konsol Function Compute, Anda dapat mengonfigurasi pemicu menggunakan metode berikut:

Gunakan Serverless Devs untuk mengonfigurasi pemicu. Untuk informasi lebih lanjut, lihat Serverless Devs.
Gunakan SDK untuk mengonfigurasi pemicu. Untuk informasi lebih lanjut, lihat SDK.

Untuk memodifikasi atau menghapus pemicu yang ada, lihat Kelola pemicu.