DataWorks:Sincronização em tempo real de uma tabela do Hologres para o Kafka

1. Selecione um tipo de tarefa de sincronização

1. Faça login no console do DataWorks. Na região desejada, clique em Data Integration > Data Integration no painel de navegação à esquerda. Escolha um workspace na lista suspensa e clique em Go to Data Integration.

2. No painel de navegação à esquerda, clique em Sync Tasks. No topo da página, clique em Create synchronization task e configure os parâmetros abaixo.

   • Data Source And Destination: Hologres→Kafka

   • New Task Name: Defina um nome personalizado para a tarefa de sincronização.

   • Synchronization Mode: single-table real-time.

   • Synchronization Mode Step: Selecione full synchronization.

2. Configure a rede e os recursos

1. Na seção Configure Network Connections and Resource Groups, selecione o Resource Group para a tarefa de sincronização. Opcionalmente, defina o Task Resource Usage em CUs.

2. Em Source Data Source, escolha uma fonte de dados Hologres. Em Destination data source, selecione uma fonte de dados Kafka. Em seguida, clique em Test Connection.

3. Após a aprovação dos testes de conectividade das fontes de dados de origem e destino, clique em Next step.

3. Configure o link de sincronização

4. Alertas

Para evitar que erros na tarefa causem atrasos na sincronização de dados de negócios, defina uma política de alertas para a tarefa de sincronização.

1. Clique em Alert Settings no canto superior direito da página para abrir a página de configurações de Alert Rule Configurations for Real-time Synchronization Subnode.

2. Clique em Add Alert Rule para configurar uma regra de alerta.

   Nota

   As regras de alerta definidas aqui se aplicam às subtarefas de sincronização em tempo real geradas por esta tarefa. Após configurar a tarefa, visualize e modifique as regras de alerta dessas subtarefas na página Run and manage real-time synchronization tasks.

3. Gerencie as regras de alerta.

   Nas regras existentes, use o botão alternador para ativá-las ou desativá-las. Também é possível direcionar alertas para diferentes destinatários com base no nível de severidade.

5. Configurações avançadas

A tarefa de sincronização oferece diversos parâmetros ajustáveis conforme a necessidade.

1. Clique em advanced settings no canto superior direito da página para abrir a página de configurações avançadas.

2. Na página advanced settings, modifique os valores dos parâmetros conforme necessário.

6. Grupo de recursos

Clique em Configure Resource Group no canto superior direito para visualizar e alterar o grupo de recursos atual da tarefa.

7. Execute a tarefa de sincronização

1. Após concluir todas as configurações, clique em Complete na parte inferior da página.

2. Na página Data Integration > Synchronization Task, localize a tarefa criada e clique em Start na coluna Operations.

3. Clique no Name/ID da tarefa correspondente na Task List para acompanhar o processo detalhado de execução.

Visualize o status da tarefa de sincronização

Após criar a tarefa de sincronização, acesse a página Tasks para visualizar todas as tarefas de sincronização do workspace e suas informações básicas.

A lista de tarefas exibe as seguintes colunas: Name/ID, Data Source Synchronization Solution (tipo de sincronização, origem e destino), Status, Execution Overview, resource group e Actions.

• Na coluna Actions, utilize as opções Start ou Stop para iniciar ou parar uma tarefa de sincronização. No menu More, execute outras operações, como Edit e View.

• Para tarefas iniciadas, acompanhe o status básico de execução no Execution Overview e clique na área de resumo correspondente para ver os detalhes da execução.

Uma tarefa de sincronização em tempo real de uma tabela do Hologres para o Kafka consiste nas três etapas a seguir:

• Structure Migration: Inclui o método de criação da tabela de destino (tabela existente ou criação automática). Se optar pela criação automática, a instrução de linguagem de definição de dados (DDL) para criar a tabela será exibida.

• Full initialization: Caso tenha selecionado Full Sync na etapa de sincronização da tarefa, o progresso da inicialização completa aparece aqui.

• Real-time Data Synchronization: Contém estatísticas sobre a sincronização em tempo real, incluindo tráfego de leitura/gravação em tempo real, dados sujos, failovers e logs de operação.

Reexecute uma tarefa

Em situações especiais, como quando é preciso modificar campos sincronizados ou ajustar informações da tabela de destino, clique em Rerun na coluna Operations da tarefa de sincronização. Essa ação sincroniza os campos ajustados e outras alterações no destino, ignorando tabelas previamente sincronizadas que não sofreram mudanças.

• Para executar a tarefa novamente sem alterações, clique em Rerun.

• Se editar a tarefa, clique em Complete após realizar as modificações. A ação da tarefa mudará para Apply Updates. Ao clicar em Apply Updates, a tarefa é reexecutada com a nova configuração.

Canal CDC

O Canal CDC é um formato de dados CDC definido pelo Alibaba Canal.

• Campos e significados

  Nome do campo	Significado do valor do campo
  id	O valor deste campo é fixo em 0.
  database	Nome do banco de dados Hologres.
  table	Nome da tabela do Hologres.
  pkNames	Coluna de chave primária da tabela do Hologres.
  isDdl	Indica se os binary logs registram alterações DDL. O valor deste campo é fixo em false, pois não há suporte para sincronização de mensagens de alterações DDL na tabela do Hologres.
  type	Tipo da alteração DML. Valores válidos: INSERT, UPDATE e DELETE. Nota Uma única operação de alteração na tabela do Hologres gera dois registros com tipo UPDATE, que serão gravados no tópico do Kafka. Um dos registros corresponde ao conteúdo dos dados antes da alteração. O outro corresponde ao conteúdo dos dados após a alteração. Para sincronizar dados completos da tabela do Hologres para o tópico do Kafka, o valor do campo type é fixo em INSERT.
  es	Timestamp de 13 dígitos em milissegundos, indicando o momento em que os dados na tabela do Hologres foram alterados. Na sincronização de dados completos da tabela do Hologres para o tópico do Kafka, o valor do campo es é fixo em 0.
  ts	Timestamp de 13 dígitos em milissegundos, indicando o momento em que a tarefa de sincronização leu os binary logs gerados para a tabela do Hologres.
  sql	Código SQL que registra alterações DDL quando os binary logs gerados para a tabela do Hologres contêm tais alterações. Como não há suporte para sincronização de mensagens de alterações DDL na tabela do Hologres, o valor deste campo é fixo como uma string vazia.
  sqlType	Tipos de dados de campo SQL aos quais os tipos de dados dos campos da tabela do Hologres correspondem. Mapeamentos entre tipos de dados do Hologres e valores válidos de sqlType: bigint: -5 decimal com escala diferente de 0: 3 decimal com escala igual a 0: -5 boolean: 16 date: 91 float4: 6 float8: 8 integer: 4 smallint: 5 json: 12 text: 12 varchar: 12 timestamp: 93 timestamptz: 93 bigserial: -5 bytea: 12 char: 12 serial: 4 time: 92 int4[]: 12 int8[]: 12 float4[]: 12 float8[]: 12 boolean[]: 12 text[]: 12
  mysqlType	Tipos de dados de campo MySQL aos quais os tipos de dados dos campos da tabela do Hologres correspondem. Mapeamentos entre tipos de dados do Hologres e valores válidos de mysqlType: bigint: BIGINT int4: INT decimal com escala diferente de 0: DECIMAL(xx,xx) decimal com escala igual a 0: BIGINT boolean: BOOLEAN date: DATE float4: FLOAT float8: DOUBLE integer: INT smallint: SMALLINT json: TEXT text: TEXT varchar: VARCHAR(xx) timestamp: DATETIME(6) timestamptz: DATETIME(6) bigserial: BIGINT bytea: TEXT char: TEXT serial: INT time: TIME(6) int4[]: TEXT int8[]: TEXT float4[]: TEXT float8[]: TEXT boolean[]: TEXT text[]: TEXT
  data	Alterações de dados na tabela do Hologres. Os nomes dos campos da tabela do Hologres são usados como chaves, e as alterações de dados nos campos são serializadas em strings e usadas como valores. Em seguida, as chaves e valores são organizados como strings formatadas em JSON. Para mais informações sobre serialização, consulte Description of JSON serialization.
  old	Uma operação de alteração na tabela do Hologres gera dois registros com tipo UPDATE, que são gravados no tópico do Kafka. Esses registros correspondem ao conteúdo dos dados antes e depois da alteração. No primeiro registro, o campo old registra o conteúdo dos dados na tabela do Hologres antes da mudança. O campo data registra o conteúdo dos dados para alterações DML que não sejam UPDATE.

• Exemplo de dados formatados em Canal JSON correspondentes a alterações de dados geradas por uma operação INSERT nos binary logs da tabela do Hologres

  {
      "id": 0,
      "database": "test",
      "table": "tp_int",
      "pkNames": [
          "id"
      ],
      "isDdl": false,
      "type": "INSERT",
      "es": 1640007049196,
      "ts": 1639633142960,
      "sql": "",
      "sqlType": {
          "bigint": -5,
          "integer": 4,
          "smallint": 5
      },
      "mysqlType": {
          "bigint": "BIGINT",
          "integer": "INT",
          "smallint": "SMALLINT"
      },
      "data": [
          {
              "bigint": "9223372036854775807",
              "integer": "2147483647",
              "smallint": "32767"
          }
      ],
      "old": null
  }

• Exemplo de dados formatados em Canal JSON correspondentes a dados completos sincronizados da tabela do Hologres

  {
      "id": 0,
      "database": "test",
      "table": "tp_int",
      "pkNames": [
          "id"
      ],
      "isDdl": false,
      "type": "INSERT",
      "es": 0,
      "ts": 1639633142960,
      "sql": "",
      "sqlType": {
          "bigint": -5,
          "integer": 4,
          "smallint": 5
      },
      "mysqlType": {
          "bigint": "BIGINT",
          "integer": "INT",
          "smallint": "SMALLINT"
      },
      "data": [
          {
              "bigint": "9223372036854775807",
              "integer": "2147483647",
              "smallint": "32767"
          }
      ],
      "old": null
  }

• Exemplo de dois registros de dados formatados em Canal JSON correspondentes a alterações de dados geradas por uma operação UPDATE nos binary logs da tabela do Hologres

  // Data content before a change
  {
      "id": 0,
      "database": "test",
      "table": "tp_int",
      "pkNames": [
          "id"
      ],
      "isDdl": false,
      "type": "UPDATE",
      "es": 1640007049196,
      "ts": 1639633142960,
      "sql": "",
      "sqlType": {
          "bigint": -5,
          "integer": 4,
          "smallint": 5
      },
      "mysqlType": {
          "bigint": "BIGINT",
          "integer": "INT",
          "smallint": "SMALLINT"
      },
      "old": [
          {
              "bigint": "0",
              "integer": "0",
              "smallint": "0"
          }
      ],
      "data": null
  }
  // Data content after a change
  {
      "id": 0,
      "database": "test",
      "table": "tp_int",
      "pkNames": [
          "id"
      ],
      "isDdl": false,
      "type": "UPDATE",
      "es": 1640007049196,
      "ts": 1639633142960,
      "sql": "",
      "sqlType": {
          "bigint": -5,
          "integer": 4,
          "smallint": 5
      },
      "mysqlType": {
          "bigint": "BIGINT",
          "integer": "INT",
          "smallint": "SMALLINT"
      },
      "data": [
          {
              "bigint": "9223372036854775807",
              "integer": "2147483647",
              "smallint": "32767"
          }
      ],
      "old": null
  }

• Exemplo de dados formatados em Canal JSON correspondentes a alterações de dados geradas por uma operação DELETE nos binary logs da tabela do Hologres

  {
      "id": 0,
      "database": "test",
      "table": "tp_int",
      "pkNames": [
          "id"
      ],
      "isDdl": false,
      "type": "DELETE",
      "es": 1640007049196,
      "ts": 1639633142960,
      "sql": "",
      "sqlType": {
          "bigint": -5,
          "integer": 4,
          "smallint": 5
      },
      "mysqlType": {
          "bigint": "BIGINT",
          "integer": "INT",
          "smallint": "SMALLINT"
      },
      "data": [
          {
              "bigint": "9223372036854775807",
              "integer": "2147483647",
              "smallint": "32767"
          }
      ],
      "old": null
  }

Json

O formato JSON utiliza os nomes dos campos dos binary logs do Hologres como chaves e serializa os valores dos campos em strings. Os pares chave-valor resultantes são gravados no tópico do Kafka como strings formatadas em JSON.