Data Transmission Service (DTS) sert de source d'événements et s'intègre à Function Compute via EventBridge. Une fois l'intégration effectuée, utilisez les déclencheurs DTS pour appeler les fonctions associées. Vous pouvez ainsi traiter des données incrémentielles en temps réel issues des tâches de suivi des modifications DTS. Cette rubrique explique comment créer un déclencheur DTS, configurer les paramètres d'entrée de la fonction, ainsi que rédiger et tester le code dans la console Function Compute.
Vue d'ensemble
Lorsque vous soumettez une demande de création de déclencheur dans la console Function Compute, ce dernier crée automatiquement des ressources flux d'événements dans EventBridge, selon la configuration du déclencheur.
Une fois les ressources créées, consultez les informations relatives au déclencheur dans la console Function Compute. Vous pouvez également visualiser les détails des ressources créées automatiquement dans la console EventBridge. Lorsqu'une tâche de suivi des modifications DTS capture des données incrémentielles depuis une base de données, elle appelle la fonction associée. Un ou plusieurs événements de message sont transmis par lots à la fonction pour traitement, conformément à vos configurations de lot.
Précautions
La tâche de suivi des modifications DTS servant de source au déclencheur doit se trouver dans la même région que la fonction Function Compute.
Si le nombre de flux d'événements créés atteint la limite, vous ne pourrez pas créer d'autres déclencheurs DTS. Pour plus d'informations sur les limites des flux d'événements, consultez la page Limites.
Prérequis
-
EventBridge
-
Function Compute
-
Data Transmission Service (DTS)
Étape 1 : Créer un déclencheur DTS
Connectez-vous à la console Function Compute. Dans le volet de navigation de gauche, choisissez .
Dans la barre de navigation supérieure, sélectionnez une région. Sur la page Functions, cliquez sur la fonction cible.
Sur la page des détails de la fonction, cliquez sur l'onglet Trigger, puis cliquez sur Create Trigger.
-
Dans le panneau Create Trigger, configurez les paramètres, puis cliquez sur OK.
Le tableau suivant décrit les paramètres de base.
Élément de configuration
Description
Exemple
Trigger Type
Le type de déclencheur. Pour plus d'informations sur les types de déclencheurs pris en charge, consultez la page Vue d'ensemble des déclencheurs.
DTS
Name
Un nom personnalisé pour le déclencheur.
dts-trigger
Version or Alias
La valeur par défaut est LATEST. Si vous souhaitez créer un déclencheur pour une autre version ou un autre alias, basculez vers cette version ou cet alias dans le coin supérieur droit de la page des détails de la fonction. Pour plus d'informations sur les versions et les alias, consultez les pages Gestion des versions et Gestion des alias.
LATEST
Tâche de suivi des modifications
Le nom de la tâche de suivi des modifications existante.
dtsqntc2***
Groupe de consommateurs
Le nom du groupe de consommateurs utilisé pour consommer les données de la tâche de suivi.
ImportantAssurez-vous que le groupe de consommateurs n'est pas exécuté sur des instances d'autres clients. Sinon, le décalage du consommateur spécifié risque de devenir invalide.
test
Compte
Le compte spécifié lors de la création du groupe de consommateurs.
test
Password
Le mot de passe spécifié lors de la création du groupe de consommateurs.
Consumer Offset
L'horodatage auquel vous souhaitez consommer le premier enregistrement de données. Le décalage du consommateur doit se situer dans la plage d'horodatage de la tâche de suivi.
RemarqueLe décalage du consommateur prend effet uniquement lorsqu'un nouveau groupe de consommateurs s'exécute pour la première fois. Si la tâche redémarre ultérieurement, la consommation reprend à partir du dernier décalage du consommateur.
2022-06-21 00:00:00
Invocation Method
Sélectionnez une méthode d'appel de fonction.
Les valeurs sont décrites ci-dessous.
Sync Invocation : Cette méthode convient aux scénarios d'appel séquentiel. Un événement unique ou un lot d'événements déclenche un appel de fonction. Le système attend que la fonction s'exécute et renvoie un résultat avant que l'événement ou le lot d'événements suivant ne déclenche un autre appel de fonction. La charge utile maximale d'une requête d'appel synchrone est de 32 Mo. Pour plus d'informations, consultez la page Appels synchrones.
Async Invocation : Cette méthode permet de consommer rapidement les événements. Un événement unique ou un lot d'événements déclenche un appel de fonction. Function Compute renvoie immédiatement une réponse, et l'événement ou le lot d'événements suivant déclenche un autre appel de fonction. Durant ce processus, la fonction est exécutée de manière asynchrone. La charge utile maximale d'une requête d'appel asynchrone est de 128 Ko. Pour plus d'informations, consultez la page Vue d'ensemble.
Appel synchrone
Trigger State
Indique si le déclencheur doit être activé immédiatement après sa création. Par défaut, l'option Enable Trigger est sélectionnée. Cela signifie que le déclencheur est activé dès sa création.
Activé
Pour plus d'informations sur les configurations avancées telles que les configurations de transmission, les nouvelles tentatives et les files d'attente de lettres mortes, consultez la page Fonctionnalités avancées.
Une fois le déclencheur créé, il apparaît dans la liste Trigger Name. Pour modifier ou supprimer le déclencheur, consultez la page Gérer les déclencheurs.
Étape 2 : Configurer les paramètres d'entrée de la fonction
La source d'événements DTS transmet un paramètre event en entrée à la fonction. Transmettez manuellement le paramètre event à la fonction afin de simuler un événement de déclenchement.
Sur l'onglet Code de la page des détails de la fonction, cliquez sur l'icône
située à côté de Test Function, puis sélectionnez Configure Test Parameters dans la liste déroulante.-
Dans le panneau Configure Test Parameters, sélectionnez Create New Test Event ou Modify Existing Test Event. Saisissez ensuite le nom et le contenu de l'événement, puis cliquez sur OK.
Voici le format de l'élément
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" } ]Pour plus d'informations sur les paramètres CloudEvents, consultez la page Vue d'ensemble des événements.
Le champ data contient les paramètres suivants.
Paramètre
Type
Description
id
String
L'ID de l'enregistrement de données DTS.
topicPartition
Array
Les informations de partition du topic.
hash
String
Le paramètre de stockage sous-jacent de DTS.
partition
String
La partition du topic.
topic
String
Le nom du topic.
offset
Int
Le décalage de stockage du message pour l'enregistrement de données DTS.
sourceTimestamp
Int
L'horodatage de génération des données DTS.
operationType
String
Le type d'opération des données DTS.
schema
Array
Le schéma de la table de base de données.
recordFields
Array
Les détails des champs.
fieldName
String
Le nom du champ.
rawDataTypeNum
Int
La valeur mappée du type de champ.
Cette valeur correspond au champ
dataTypeNumberdans les données incrémentielles désérialisées provenant du canal de suivi des modifications. Pour plus d'informations, consultez la page Utiliser un client Kafka pour consommer les données suivies.isPrimaryKey
Boolean
Indique si le champ est une clé primaire.
isUniqueKey
Boolean
Indique si le champ est une clé unique.
fieldPosition
String
La position du champ.
nameIndex
Array
L'index nommé.
schemaId
String
L'ID du schéma de la table de base de données.
databaseName
String
Le nom de la base de données.
tableName
String
Le nom de la table.
primaryIndexInfo
String
L'index de clé primaire.
indexType
String
Le type d'index de clé primaire.
indexFields
Array
Les champs de l'index de clé primaire.
cardinality
String
La cardinalité de la clé primaire.
nullable
Boolean
Indique si la clé primaire peut être nulle.
isFirstUniqueIndex
Boolean
Indique s'il s'agit du premier index unique.
uniqueIndexInfo
String
L'index unique.
foreignIndexInfo
String
L'index de clé étrangère.
normalIndexInfo
String
L'index normal.
databaseInfo
Array
Les informations relatives à la base de données.
databaseType
String
Le type de base de données.
version
String
La version de la base de données.
totalRows
Int
Le nombre total de lignes dans la table.
beforeImage
String
L'état des champs d'enregistrement avant l'opération.
values
String
Les valeurs des champs d'enregistrement.
size
Int
La taille des champs d'enregistrement.
afterImage
String
L'état des champs d'enregistrement après l'opération.
Étape 3 : Rédiger et tester le code de la fonction
Après avoir créé le déclencheur, rédigez et testez le code de la fonction pour vérifier son exactitude. En pratique, lorsque la tâche de suivi des modifications DTS capture des données incrémentielles depuis la base de données, le déclencheur appelle automatiquement la fonction.
-
Sur l'onglet Code de la page des détails de la fonction, rédigez le code dans l'éditeur de code, puis cliquez sur Deploy.
Cette rubrique utilise du code Node.js à titre d'exemple.
'use strict'; /* To enable the initializer feature, implement the initializer function as follows: exports.initializer = (context, callback) => { console.log('initializing'); callback(null, ''); }; */ exports.handler = (event, context, callback) => { console.log("event: %s", event); // Parse the event parameter and process the event. callback(null, 'return result'); } Cliquez sur Test Function.
Informations complémentaires
Outre la console Function Compute, vous pouvez également configurer des déclencheurs de différentes manières :
Utilisez l'outil Serverless Devs pour configurer les déclencheurs. Pour plus d'informations, consultez la page Commandes courantes de Serverless Devs.
Utilisez les SDK pour configurer les déclencheurs. Pour plus d'informations, consultez la page SDK.
Pour modifier ou supprimer un déclencheur, consultez la page Gérer les déclencheurs.