Cara mengelola pemicu DTS di konsol - Function Compute - Alibaba Cloud Documentation Center

Data Transmission Service (DTS) berperan sebagai sumber event dan terintegrasi dengan Function Compute melalui EventBridge. Setelah integrasi, Anda dapat menggunakan pemicu DTS untuk memanggil fungsi terkait, sehingga memungkinkan pemrosesan kustom terhadap data inkremental real-time yang diambil dari tugas pelacakan perubahan DTS. Topik ini menjelaskan cara membuat pemicu DTS, mengonfigurasi parameter input fungsi, serta menulis dan menguji kode di konsol Function Compute.

Ikhtisar

Setelah Anda mengirim permintaan untuk membuat pemicu di konsol Function Compute, Function Compute secara otomatis membuat sumber daya event stream di EventBridge berdasarkan konfigurasi pemicu tersebut.

Setelah sumber daya dibuat, Anda dapat melihat informasi pemicu di konsol Function Compute serta informasi mengenai sumber daya yang dibuat secara otomatis di konsol EventBridge. Saat tugas pelacakan perubahan DTS menangkap data inkremental dari database, fungsi terkait akan dipanggil. Satu atau beberapa event pesan didorong ke fungsi dalam bentuk batch untuk diproses sesuai dengan konfigurasi batch Anda.

Perhatian

Tugas pelacakan perubahan DTS yang menjadi sumber pemicu harus berada di wilayah yang sama dengan fungsi Function Compute.
Jika jumlah event stream yang dibuat telah mencapai batas, Anda tidak dapat membuat pemicu DTS tambahan. Untuk informasi lebih lanjut mengenai batas event stream, lihat Batas.

Prasyarat

EventBridge
- Aktifkan EventBridge dan berikan izin
Function Compute
- Buat fungsi
Data Transmission Service (DTS)
- Buat tugas pelacakan perubahan
- Buat kelompok konsumen

Langkah 1: Buat pemicu DTS

Masuk ke konsol Function Compute. Di panel navigasi sebelah kiri, pilih Function Management > Functions.
Di bilah navigasi atas, pilih wilayah. Pada halaman Functions, klik fungsi yang dituju.
Di halaman detail fungsi, klik tab Trigger, lalu klik Create Trigger.

Di panel Create Trigger, konfigurasikan parameter, lalu klik OK.

Tabel berikut menjelaskan parameter dasar.

Item Konfigurasi	Deskripsi	Contoh
Trigger Type	Jenis pemicu. Untuk informasi selengkapnya mengenai jenis pemicu yang didukung, lihat Ikhtisar pemicu.	DTS
Name	Nama kustom untuk pemicu.	dts-trigger
Version or Alias	Nilai default adalah LATEST. Jika ingin membuat pemicu untuk versi atau alias lain, alihkan ke versi atau alias tersebut di pojok kanan atas halaman detail fungsi. Untuk informasi selengkapnya mengenai versi dan alias, lihat Manajemen versi dan Manajemen alias.	LATEST
Change Tracking Task	Nama tugas pelacakan perubahan yang sudah ada.	dtsqntc2***
Consumer Group	Nama kelompok konsumen yang digunakan untuk mengonsumsi data dari tugas pelacakan. Penting Pastikan kelompok konsumen tidak sedang berjalan di instans client lain. Jika tidak, offset konsumen yang ditentukan mungkin menjadi tidak valid.	test
Account	Akun yang ditentukan saat membuat kelompok konsumen.	test
Password	Kata sandi yang ditentukan saat membuat kelompok konsumen.	******
Consumer Offset	Waktu ketika Anda ingin mengonsumsi catatan data pertama. Offset konsumen harus berada dalam rentang waktu tugas pelacakan. Catatan Offset konsumen hanya berlaku saat kelompok konsumen baru pertama kali dijalankan. Jika tugas dijalankan ulang nanti, konsumsi dilanjutkan dari offset konsumen terakhir.	2022-06-21 00:00:00
Invocation Method	Pilih metode pemanggilan fungsi. Berikut penjelasan nilainya. Sync Invocation: Metode ini cocok untuk skenario pemanggilan berurutan. Satu event atau batch event memicu pemanggilan fungsi. Sistem menunggu fungsi dieksekusi dan mengembalikan hasil sebelum event atau batch event berikutnya memicu pemanggilan fungsi lainnya. Muatan maksimum permintaan pemanggilan sinkron adalah 32 MB. Untuk informasi selengkapnya, lihat Pemanggilan sinkron. Async Invocation: Metode ini memungkinkan Anda mengonsumsi event dengan cepat. Satu event atau batch event memicu pemanggilan fungsi. Function Compute langsung mengembalikan respons, dan event atau batch event berikutnya memicu pemanggilan fungsi lainnya. Selama proses ini, fungsi dieksekusi secara asinkron. Muatan maksimum permintaan pemanggilan asinkron adalah 128 KB. Untuk informasi selengkapnya, lihat Ikhtisar.	Synchronous Invocation
Trigger State	Menentukan apakah pemicu diaktifkan segera setelah dibuat. Secara default, opsi Enable Trigger dipilih. Artinya, pemicu langsung diaktifkan setelah dibuat.	Enabled

Untuk informasi selengkapnya mengenai konfigurasi lanjutan seperti konfigurasi dorong, percobaan ulang, dan antrian surat mati (dead-letter queue), lihat Fitur lanjutan.

Setelah pemicu dibuat, pemicu tersebut muncul di daftar Trigger Name. Untuk mengubah atau menghapus pemicu, lihat Kelola pemicu.

Langkah 2: Konfigurasi parameter input fungsi

Sumber event DTS meneruskan event sebagai parameter input ke fungsi. Anda dapat secara manual meneruskan event ke fungsi untuk mensimulasikan peristiwa pemicu.

Di tab Code halaman detail fungsi, klik ikon di samping Test Function dan pilih Configure Test Parameters dari daftar drop-down.

Di panel Configure Test Parameters, pilih Create New Test Event atau Modify Existing Test Event. Lalu, masukkan nama dan konten event, kemudian klik OK.

Berikut adalah format 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"
  }
]

Tabel berikut menjelaskan parameter dalam bidang data.

Parameter	Tipe	Deskripsi
id	String	ID data DTS.
topicPartition	Array	Informasi mengenai partisi topik.
hash	String	Parameter penyimpanan dasar DTS.
partition	String	Partisi dari topik.
topic	String	Nama topik.
offset	Int	Menentukan offset log broker untuk data DTS.
sourceTimestamp	Int	Waktu saat data DTS dihasilkan.
operationType	String	Jenis operasi data DTS.
schema	Array	Informasi mengenai skema tabel database.
recordFields	Array	Detail catatan bidang.
fieldName	String	Nama bidang.
rawDataTypeNum	Int	Nilai pemetaan tipe bidang. Nilai ini sesuai dengan nilai bidang `dataTypeNumber` setelah data inkremental yang diperoleh dari saluran pelacakan perubahan dideserialisasi. Untuk informasi selengkapnya, lihat Gunakan client Kafka untuk mengonsumsi data yang berlangganan.
isPrimaryKey	Boolean	Menunjukkan apakah bidang tersebut merupakan kunci primer.
isUniqueKey	Boolean	Menunjukkan apakah bidang tersebut memiliki nilai unik.
fieldPosition	String	Posisi bidang.
nameIndex	Array	Indeks bernama.
schemaId	String	ID informasi skema tabel database.
databaseName	String	Nama database.
tableName	String	Nama tabel data.
primaryIndexInfo	String	Indeks kunci primer.
indexType	String	Jenis indeks kunci primer.
indexFields	Array	Konten bidang indeks kunci primer.
cardinality	String	Kardinalitas kunci primer.
nullable	Boolean	Menunjukkan apakah kunci primer dapat bernilai null.
isFirstUniqueIndex	Boolean	Menunjukkan apakah ini adalah indeks unik pertama.
uniqueIndexInfo	String	Indeks unik.
foreignIndexInfo	String	Indeks kunci asing.
normalIndexInfo	String	Indeks normal.
databaseInfo	Array	Informasi mengenai database.
databaseType	String	Jenis database.
version	String	Versi database.
totalRows	Int	Jumlah total baris dalam tabel data.
beforeImage	String	Gambar konten bidang catatan sebelum operasi.
values	String	Nilai bidang catatan.
size	Int	Ukuran bidang catatan.
afterImage	String	Gambar konten bidang catatan setelah operasi.

Langkah 3: Tulis dan uji kode fungsi

Setelah membuat pemicu, Anda dapat menulis dan menguji kode fungsi untuk memverifikasi kebenarannya. Dalam praktiknya, ketika tugas pelacakan perubahan DTS menangkap data inkremental dari database, pemicu secara otomatis memanggil fungsi tersebut.

Di tab Code halaman detail fungsi, Anda dapat menulis kode di editor kode dan klik Deploy.

Topik ini menggunakan kode Node.js sebagai contoh.

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

Klik Test Function.

Informasi lebih lanjut

Selain melalui konsol Function Compute, Anda juga dapat mengonfigurasi pemicu dengan cara berikut:

Gunakan tool Serverless Devs untuk mengonfigurasi pemicu. Untuk informasi selengkapnya, lihat Perintah umum Serverless Devs.
Gunakan SDK untuk mengonfigurasi pemicu. Untuk informasi selengkapnya, lihat SDK.

Untuk mengubah atau menghapus pemicu, lihat Kelola pemicu.