Como gerenciar gatilhos DTS no console - Function Compute: Central de Documentação do Alibaba Cloud

Visão geral

Ao envie uma solicitação de criação de gatilho no console do Function Compute, o serviço cria automaticamente recursos de fluxo de eventos no EventBridge com base na configuração do gatilho.

Após a criação dos recursos, visualize as informações do gatilho no console do Function Compute. Também é possível consultar os detalhes dos recursos criados automaticamente no console do EventBridge. Quando uma tarefa de rastreamento de alterações do DTS captura dados incrementais de um banco de dados, a função associada é invocada. Um ou mais eventos de mensagem são enviados em lotes para a função, conforme suas configurações de lote, para processamento.

Precauções

A tarefa de rastreamento de alterações do DTS que serve como origem do gatilho deve estar na mesma região da função do Function Compute.
Se o número de fluxos de eventos criados atingir o limite, não será possível criar novos gatilhos DTS. Para mais informações sobre o limite de fluxos de eventos, consulte Limites.

Pré-requisitos

EventBridge
- Ativar o EventBridge e conceder permissões
Function Compute
- Criar uma função
Data Transmission Service (DTS)
- Criar uma tarefa de rastreamento de alterações
- Criar um grupo de consumidores

Etapa 1: Criar um gatilho DTS

Faça login no console do Function Compute. No painel de navegação à esquerda, escolha Function Management > Functions.
Na barra de navegação superior, selecione uma região. Na página Functions, clique em na função desejada.
Na página de detalhes da função, clique em na aba Trigger e clique em Create Trigger.

No painel Create Trigger, configure os parâmetros e clique em OK.

A tabela a seguir descreve os parâmetros básicos.

Item de Configuração	Descrição	Exemplo
Trigger Type	Tipo do gatilho. Para mais informações sobre os tipos de gatilho compatíveis, consulte Visão geral de gatilhos.	DTS
Name	Nome personalizado para o gatilho.	dts-trigger
Version or Alias	O valor padrão é LATEST. Para criar um gatilho para outra versão ou alias, alterne para essa versão ou alias no canto superior direito da página de detalhes da função. Para mais informações sobre versões e aliases, consulte Gerenciamento de versões e Gerenciamento de aliases.	LATEST
Change Tracking Task	Nome da tarefa de rastreamento de alterações existente.	dtsqntc2***
Consumer Group	Nome do grupo de consumidores usado para consumir dados da tarefa de rastreamento. Importante Certifique-se de que o grupo de consumidores não esteja em execução em instâncias de outros clientes. Caso contrário, o offset de consumo especificado pode se tornar inválido.	test
Account	Conta especificada durante a criação do grupo de consumidores.	test
Password	Senha especificada durante a criação do grupo de consumidores.
Consumer Offset	Timestamp a partir do qual você deseja consumir o primeiro registro de dados. O offset de consumo deve estar dentro do intervalo de timestamps da tarefa de rastreamento. Nota O offset de consumo só entra em vigor quando um novo grupo de consumidores é executado pela primeira vez. Se a tarefa for reiniciada posteriormente, o consumo continuará a partir do último offset registrado.	2022-06-21 00:00:00
Invocation Method	Selecione um método de invocação da função. Os valores são descritos abaixo. Sync Invocation: Adequado para cenários de invocação sequencial. Um único evento ou um lote de eventos aciona a invocação da função. O sistema aguarda a execução da função e o retorno de um resultado antes que o próximo evento ou lote acione outra invocação. A carga útil máxima de uma solicitação de invocação síncrona é de 32 MB. Para mais informações, consulte Invocações síncronas. Async Invocation: Permite consumir eventos rapidamente. Um único evento ou lote aciona a invocação da função. O Function Compute retorna uma resposta imediatamente, e o próximo evento ou lote já pode acionar outra invocação. Durante esse processo, a função é executada de forma assíncrona. A carga útil máxima de uma solicitação de invocação assíncrona é de 128 KB. Para mais informações, consulte Visão geral.	Synchronous Invocation
Trigger State	Define se o gatilho deve ser ativado imediatamente após a criação. Por padrão, a opção Enable Trigger vem selecionada, o que significa que o gatilho é ativado assim que criado.	Enabled

Para mais informações sobre configurações avançadas, como configurações de envio, tentativas e filas de mensagens mortas, consulte Recursos avançados.

Após a criação, o gatilho aparece na lista Trigger Name. Para modifique ou exclua o gatilho, consulte Gerenciar gatilhos.

Etapa 2: Configure parâmetros de entrada da função

A fonte de eventos DTS passa um event como parâmetro de entrada para a função. Passe manualmente o event para a função a fim de simular um evento de acionamento.

Na aba Code da página de detalhes da função, clique em no ícone ao lado de Test Function e selecione Configure Test Parameters na lista suspensa.

No painel Configure Test Parameters, selecione Create New Test Event ou Modify Existing Test Event. Em seguida, insira o nome e o conteúdo do evento e clique em OK.

O formato do event é o seguinte:

[
  {
    "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 descrições dos parâmetros na especificação CloudEvents, consulte Visão geral de eventos.

Os parâmetros no campo data estão descritos na tabela a seguir.

Parâmetro	Tipo	Descrição
id	String	ID do registro de dados do DTS.
topicPartition	Array	Informações de partição do tópico.
hash	String	Parâmetro de armazenamento subjacente do DTS.
partition	String	Partição do tópico.
topic	String	Nome do tópico.
offset	Int	Offset de armazenamento da mensagem para o registro de dados do DTS.
sourceTimestamp	Int	Timestamp de geração dos dados do DTS.
operationType	String	Tipo de operação dos dados do DTS.
schema	Array	Schema da tabela do banco de dados.
recordFields	Array	Detalhes dos campos.
fieldName	String	Nome do campo.
rawDataTypeNum	Int	Valor mapeado do tipo de campo. Este valor corresponde ao campo `dataTypeNumber` nos dados incrementais desserializados obtidos do canal de rastreamento de alterações. Para mais informações, consulte Usar um cliente Kafka para consumir dados rastreados.
isPrimaryKey	Boolean	Indica se o campo é uma chave primária.
isUniqueKey	Boolean	Indica se o campo é uma chave única.
fieldPosition	String	Posição do campo.
nameIndex	Array	Índice nomeado.
schemaId	String	ID do schema da tabela do banco de dados.
databaseName	String	Nome do banco de dados.
tableName	String	Nome da tabela.
primaryIndexInfo	String	Índice de chave primária.
indexType	String	Tipo do índice de chave primária.
indexFields	Array	Campos no índice de chave primária.
cardinality	String	Cardinalidade da chave primária.
nullable	Boolean	Indica se a chave primária pode ser nula.
isFirstUniqueIndex	Boolean	Indica se é o primeiro índice único.
uniqueIndexInfo	String	Índice único.
foreignIndexInfo	String	Índice de chave estrangeira.
normalIndexInfo	String	Índice normal.
databaseInfo	Array	Informações sobre o banco de dados.
databaseType	String	Tipo do banco de dados.
version	String	Versão do banco de dados.
totalRows	Int	Número total de linhas na tabela.
beforeImage	String	Estado dos campos do registro antes da operação.
values	String	Valores dos campos do registro.
size	Int	Tamanho dos campos do registro.
afterImage	String	Estado dos campos do registro após a operação.

Etapa 3: Escrever e testar o código da função

Após criar o gatilho, escreva e teste o código da função para verifique sua correção. Na prática, quando a tarefa de rastreamento de alterações do DTS captura dados incrementais do banco de dados, o gatilho invoca a função automaticamente.

Na aba Code da página de detalhes da função, escreva o código no editor e clique em Deploy.

Este tópico usa código Node.js como exemplo.

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

Clique em Test Function.

Mais informações

Além do console do Function Compute, configure gatilhos das seguintes formas:

Use a ferramenta Serverless Devs para configure gatilhos. Para mais informações, consulte Comandos comuns do Serverless Devs.
Utilize SDKs para configure gatilhos. Para mais informações, consulte SDKs.

Para modifique ou exclua um gatilho, consulte Gerenciar gatilhos.