How to manage DTS triggers in the console - Function Compute - Centro de documentación de Alibaba Cloud

Data Transmission Service (DTS) actúa como origen de eventos y se integra con Function Compute mediante EventBridge. Una vez configurada la integración, los disparadores de DTS permiten invocar las funciones asociadas. De este modo, es posible aplicar un procesamiento personalizado a los datos incrementales en tiempo real que se obtienen de las tareas de seguimiento de cambios de DTS. En este tema se explica cómo crear un disparador de DTS, configurar los parámetros de entrada de la función, y escribir y probar el código en la consola de Function Compute.

Descripción general

Cuando usted envía una solicitud para crear un disparador en la consola de Function Compute, el servicio genera automáticamente los recursos event stream en EventBridge según la configuración del disparador.

Una vez creados los recursos, consulte la información del disparador en la consola de Function Compute. También puede revisar los detalles de los recursos generados automáticamente en la consola de EventBridge. Cuando una tarea de seguimiento de cambios de DTS captura datos incrementales de una base de datos, se invoca la función asociada. Los eventos de mensaje se envían a la función por lotes para su procesamiento, de acuerdo con la configuración de lotes establecida.

Precauciones

La tarea de seguimiento de cambios de DTS que sirve como origen del disparador debe ubicarse en la misma región que la función de Function Compute.
Si la cantidad de flujos de eventos creados alcanza el límite, no podrá crear más disparadores de DTS. Para obtener más información sobre el límite de flujos de eventos, consulte Limits.

Requisitos previos

EventBridge
- Activate EventBridge and grant permissions
Function Compute
- Create a function
Data Transmission Service (DTS)
- Create a change tracking task
- Create a consumer group

Paso 1: Crear un disparador de DTS

Inicie sesión en la consola de Function Compute. En el panel de navegación izquierdo, seleccione Function Management > Functions.
En la barra de navegación superior, seleccione una región. En la página Functions, haga clic en la función de destino.
En la página de detalles de la función, haga clic en la pestaña Trigger y luego en Create Trigger.

En el panel Create Trigger, configure los parámetros y haga clic en OK.

A continuación se describen los parámetros básicos.

Elemento de configuración	Descripción	Ejemplo
Trigger Type	El tipo de disparador. Para obtener más información sobre los tipos de disparador compatibles, consulte Trigger overview.	DTS
Name	Un nombre personalizado para el disparador.	dts-trigger
Version or Alias	El valor predeterminado es LATEST. Si desea crear un disparador para otra versión o alias, cambie a dicha versión o alias en la esquina superior derecha de la página de detalles de la función. Para obtener más información sobre versiones y alias, consulte Version management y Alias management.	LATEST
Change Tracking Task	El nombre de la tarea de seguimiento de cambios existente.	dtsqntc2***
Consumer Group	El nombre del grupo de consumidores que se utiliza para consumir datos de la tarea de seguimiento. Importante Compruebe que el grupo de consumidores no se esté ejecutando en instancias de otros clientes. De lo contrario, el desplazamiento de consumo especificado podría dejar de ser válido.	test
Account	La cuenta que se especificó al crear el grupo de consumidores.	test
Password	La contraseña que se especificó al crear el grupo de consumidores.
Consumer Offset	La marca de tiempo en la que desea consumir el primer registro de datos. El desplazamiento de consumo debe estar dentro del rango de marcas de tiempo de la tarea de seguimiento. Nota El desplazamiento de consumo solo surte efecto cuando un grupo de consumidores nuevo se ejecuta por primera vez. Si la tarea se reinicia después, el consumo continúa desde el último desplazamiento registrado.	2022-06-21 00:00:00
Invocation Method	Seleccione un método de invocación de la función. A continuación se describen los valores disponibles. Sync Invocation: este método es adecuado para escenarios de invocación secuencial. Un evento individual o un lote de eventos activa la invocación de la función. El sistema espera a que la función se ejecute y devuelva un resultado antes de que el siguiente evento o lote active otra invocación. La carga útil máxima de una solicitud de invocación síncrona es de 32 MB. Para obtener más información, consulte Synchronous invocations. Async Invocation: esta opción permite consumir eventos rápidamente. Un evento individual o un lote de eventos activa la invocación de la función. Function Compute devuelve una respuesta de inmediato y el siguiente evento o lote desencadena otra invocación. Durante este proceso, la función se ejecuta de forma asíncrona. La carga útil máxima de una solicitud de invocación asíncrona es de 128 KB. Para obtener más información, consulte Overview.	Synchronous Invocation
Trigger State	Indica si se habilita el disparador inmediatamente después de crearlo. De forma predeterminada, la opción Enable Trigger está seleccionada. Esto significa que el disparador se habilita justo después de su creación.	Enabled

Para obtener más información sobre configuraciones avanzadas, como los ajustes de envío, los reintentos y las colas de mensajes no entregados, consulte Advanced features.

Una vez creado el disparador, este aparece en la lista Trigger Name. Para modificar o eliminar el disparador, consulte Manage triggers.

Paso 2: Configurar los parámetros de entrada de la función

El origen de eventos de DTS pasa un event como parámetro de entrada a la función. Puede pasar manualmente el event a la función para simular un evento de activación.

En la pestaña Code de la página de detalles de la función, haga clic en el icono junto a Test Function y seleccione Configure Test Parameters en la lista desplegable.

En el panel Configure Test Parameters, seleccione Create New Test Event o Modify Existing Test Event. A continuación, introduzca el nombre y el contenido del evento, y haga clic en OK.

El formato del event es el siguiente:

[
  {
    "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"
  }
]

Para obtener más información sobre los parámetros de CloudEvents, consulte Event overview.

El campo data contiene los siguientes parámetros.

Parámetro	Tipo	Descripción
`id`	String	El identificador del registro de datos de DTS.
`topicPartition`	Array	La información de partición del tema.
`hash`	String	El parámetro de almacenamiento subyacente de DTS.
`partition`	String	La partición del tema.
`topic`	String	El nombre del tema.
`offset`	Int	El desplazamiento de almacenamiento del mensaje para el registro de datos de DTS.
`sourceTimestamp`	Int	La marca de tiempo en la que se generaron los datos de DTS.
`operationType`	String	El tipo de operación de los datos de DTS.
`schema`	Array	El esquema de la tabla de la base de datos.
`recordFields`	Array	Los detalles de los campos.
`fieldName`	String	El nombre del campo.
`rawDataTypeNum`	Int	El valor asignado al tipo de campo. Este valor corresponde al campo `dataTypeNumber` en los datos incrementales deserializados del canal de seguimiento de cambios. Para obtener más información, consulte Use a Kafka client to consume tracked data.
`isPrimaryKey`	Boolean	Indica si el campo es una clave principal.
`isUniqueKey`	Boolean	Indica si el campo es una clave única.
`fieldPosition`	String	La posición del campo.
`nameIndex`	Array	El índice con nombre.
`schemaId`	String	El identificador del esquema de la tabla de la base de datos.
`databaseName`	String	El nombre de la base de datos.
`tableName`	String	El nombre de la tabla.
`primaryIndexInfo`	String	El índice de clave principal.
`indexType`	String	El tipo de índice de clave principal.
`indexFields`	Array	Los campos del índice de clave principal.
`cardinality`	String	La cardinalidad de la clave principal.
`nullable`	Boolean	Indica si la clave principal admite valores nulos.
`isFirstUniqueIndex`	Boolean	Indica si se trata del primer índice único.
`uniqueIndexInfo`	String	El índice único.
`foreignIndexInfo`	String	El índice de clave externa.
`normalIndexInfo`	String	El índice normal.
`databaseInfo`	Array	La información de la base de datos.
`databaseType`	String	El tipo de base de datos.
`version`	String	La versión de la base de datos.
`totalRows`	Int	El número total de filas de la tabla.
`beforeImage`	String	El estado de los campos del registro antes de la operación.
`values`	String	Los valores de los campos del registro.
`size`	Int	El tamaño de los campos del registro.
`afterImage`	String	El estado de los campos del registro después de la operación.

Paso 3: Escribir y probar el código de la función

Después de crear el disparador, escriba y pruebe el código de la función para comprobar que funciona correctamente. En la práctica, cuando la tarea de seguimiento de cambios de DTS captura datos incrementales de la base de datos, el disparador invoca la función automáticamente.

En la pestaña Code de la página de detalles de la función, escriba el código en el editor y haga clic en Deploy.

En este tema se usa código de Node.js como ejemplo.

'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');
}

Haga clic en Test Function.

Más información

Además de la consola de Function Compute, también puede configurar disparadores de las siguientes maneras:

Use la herramienta Serverless Devs para configurar disparadores. Para obtener más información, consulte Common commands of Serverless Devs.
Utilice los SDK para configurar disparadores. Para obtener más información, consulte SDKs.

Para modificar o eliminar un disparador, consulte Manage triggers.