Cara mengelola pemicu DTS di konsol - Function Compute

Data Transmission Service (DTS) adalah sumber event yang terintegrasi dengan Function Compute melalui EventBridge. Integrasi ini memungkinkan pemicu DTS untuk memanggil fungsi terkait guna memproses data inkremental secara real-time dari tugas pelacakan perubahan database. Topik ini menjelaskan cara membuat pemicu DTS, mengonfigurasi parameter input, serta menulis dan menguji kode di konsol Function Compute.

Ikhtisar

Saat Anda membuat pemicu di konsol Function Compute, Function Compute secara otomatis membuat event streams di EventBridge berdasarkan konfigurasi pemicu tersebut.

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

Perhatian

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

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 Triggers, lalu klik Create Trigger.

Di panel Buat Pemicu, konfigurasikan parameter, lalu klik OK.

Tabel berikut menjelaskan item konfigurasi dasar.

Item Konfigurasi	Deskripsi	Contoh
Trigger Type	Jenis pemicu. Untuk informasi selengkapnya tentang jenis pemicu yang didukung, lihat Ikhtisar pemicu.	DTS
Name	Nama kustom untuk pemicu.	dts-trigger
Version Or Alias	Nilai default-nya adalah LATEST. Untuk membuat pemicu pada versi atau alias lain, alihkan terlebih dahulu ke versi atau alias tersebut di pojok kanan atas halaman detail fungsi. Untuk informasi selengkapnya tentang versi dan alias, lihat Mengelola versi dan Mengelola alias.	LATEST
Change Tracking Task	Nama tugas pelacakan perubahan yang sudah ada.	dtsqntc2***
Consumer Group	Nama kelompok konsumen yang dibuat untuk mengonsumsi data dari tugas pelacakan. Penting Pastikan kelompok konsumen tidak sedang berjalan di instans klien lain. Jika tidak, offset konsumen yang ditentukan mungkin menjadi tidak valid.	test
Account	Akun yang Anda tetapkan saat membuat kelompok konsumen.	test
Password	Kata sandi yang Anda tetapkan saat membuat kelompok konsumen.	******
Consumer Offset	Waktu ketika Anda ingin mulai mengonsumsi catatan data pertama. Offset konsumen harus berada dalam rentang waktu instansi langganan. Catatan Offset konsumen hanya berlaku saat kelompok konsumen baru dijalankan pertama kali. Jika tugas dijalankan ulang nanti, konsumsi dilanjutkan dari offset konsumen terakhir.	2022-06-21 00:00:00
Invocation Method	Pilih metode untuk memanggil fungsi. Nilai yang valid dijelaskan sebagai berikut: Synchronous Call: 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 panggilan sinkron adalah 32 MB. Untuk informasi selengkapnya, lihat Panggilan sinkron. Asynchronous Invocation: Metode ini memungkinkan Anda mengonsumsi event dengan cepat. Satu event atau batch event memicu pemanggilan fungsi. Function Compute langsung mengembalikan tanggapan, dan event atau batch event berikutnya kemudian dapat memicu pemanggilan fungsi lainnya. Selama proses ini, fungsi dieksekusi secara asinkron. Muatan maksimum permintaan pemanggilan asinkron adalah 128 KB. Untuk informasi selengkapnya, lihat Ikhtisar.	Panggilan Sinkron
Trigger State	Menentukan apakah pemicu diaktifkan segera setelah dibuat. Secara default, Enable Trigger dipilih, artinya pemicu langsung diaktifkan setelah dibuat.	Diaktifkan

Untuk informasi selengkapnya tentang item konfigurasi lanjutan, seperti konfigurasi dorong, percobaan ulang, dan antrian pesan gagal, lihat Fitur lanjutan.

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

Langkah 2: Konfigurasi parameter input fungsi

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

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

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

event menggunakan format berikut:

[
  {
    "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 mengenai parameter yang didefinisikan dalam spesifikasi CloudEvents, lihat Ikhtisar event.

Tabel berikut menjelaskan parameter yang terdapat dalam bidang data.

Parameter	Tipe	Deskripsi
id	String	ID entri data DTS.
topicPartition	Array	Informasi partisi mengenai topik tempat event didorong.
hash	String	Parameter penyimpanan dasar DTS.
partition	String	Partisi.
topic	String	Nama topik.
offset	Int	Offset entri data DTS.
sourceTimestamp	Int	Waktu pembuatan entri data DTS.
operationType	String	Jenis operasi yang terlibat dalam entri data DTS.
schema	Array	Informasi skema mengenai database.
recordFields	Array	Rincian bidang.
fieldName	String	Nama bidang.
rawDataTypeNum	Int	Nilai pemetaan tipe bidang. Nilai parameter ini sesuai dengan nilai field dataTypeNumber dalam data inkremental yang telah dideserialisasi dari instansi pelacakan perubahan. Untuk informasi selengkapnya, lihat Gunakan klien Kafka untuk mengonsumsi data yang dilacak.
isPrimaryKey	Boolean	Menunjukkan apakah bidang tersebut merupakan bidang kunci primer.
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 database.
databaseName	String	Nama database.
tableName	String	Nama tabel.
primaryIndexInfo	String	Indeks kunci primer.
indexType	String	Jenis indeks.
indexFields	Array	Bidang tempat indeks dibuat.
cardinality	String	Kardinalitas kunci primer.
nullable	Boolean	Menunjukkan apakah kunci primer dapat bernilai null.
isFirstUniqueIndex	Boolean	Menunjukkan apakah indeks tersebut merupakan indeks unik pertama.
uniqueIndexInfo	String	Indeks unik.
foreignIndexInfo	String	Indeks untuk kunci asing.
normalIndexInfo	String	Indeks reguler.
databaseInfo	Array	Informasi mengenai database.
databaseType	String	Engine database.
version	String	Versi engine database.
totalRows	Int	Jumlah total baris dalam tabel.
beforeImage	String	Citra yang merekam nilai bidang sebelum operasi dilakukan.
values	String	Nilai bidang yang direkam.
size	Int	Ukuran bidang yang direkam.
afterImage	String	Citra yang merekam nilai bidang setelah operasi dilakukan.

Langkah 3: Tulis dan uji kode fungsi

Setelah membuat pemicu, Anda dapat menulis dan menguji kode fungsi. Saat tugas pelacakan perubahan DTS menangkap data inkremental dari database, pemicu secara otomatis memanggil fungsi tersebut.

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

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);
  // Parsing parameter event dan proses event.
  callback(null, 'return result');
}

Klik Test Function.

Informasi lebih lanjut

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

Gunakan alat 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 Mengelola pemicu.