DataWorks:Sincronización en tiempo real de una tabla de Hologres a Kafka

1. Seleccione un tipo de tarea de sincronización

1. Inicie sesión en la consola de DataWorks. En la región objetivo, haga clic en Data Integration > Data Integration en el panel de navegación izquierdo. Seleccione un espacio de trabajo en la lista desplegable y haga clic en Go to Data Integration.

2. En el panel de navegación izquierdo, haga clic en Sync Tasks. En la parte superior de la página, haga clic en Create synchronization task y configure los siguientes parámetros.

   • Data Source And Destination: Hologres→Kafka

   • New Task Name: Personalice un nombre para la tarea de sincronización.

   • Synchronization Mode: single-table real-time.

   • Synchronization Mode Step: Seleccione full synchronization.

2. Configure la red y los recursos

1. En la sección Configure Network Connections and Resource Groups, seleccione el Resource Group para la tarea de sincronización. También puede establecer el Task Resource Usage en CUs.

2. Para el Source Data Source, seleccione un origen de datos de Hologres. Para el Destination data source, seleccione un origen de datos de Kafka. A continuación, haga clic en Test Connection.

3. Una vez que las pruebas de conectividad para los orígenes de datos de origen y destino sean satisfactorias, haga clic en Next step.

3. Configure el vínculo de sincronización

4. Alertas

Para evitar que los errores de la tarea provoquen retrasos en la sincronización de datos empresariales, puede establecer una directiva de alertas para la tarea de sincronización.

1. Haga clic en Alert Settings en la esquina superior derecha de la página para abrir la página de configuración de Alert Rule Configurations for Real-time Synchronization Subnode.

2. Haga clic en Add Alert Rule para configurar una regla de alerta.

   Nota

   Las reglas de alerta que defina aquí se aplican a las subtareas de sincronización en tiempo real que genera esta tarea. Después de configurar la tarea, puede consultar y modificar las reglas de alerta para estas subtareas en la página Run and manage real-time synchronization tasks.

3. Gestione las reglas de alerta.

   Para las reglas de alerta existentes, puede usar el interruptor para habilitarlas o deshabilitarlas. También puede enviar alertas a diferentes destinatarios según el nivel de alerta.

5. Configuración avanzada

La tarea de sincronización proporciona varios parámetros que puede modificar según sea necesario.

1. Haga clic en advanced settings en la esquina superior derecha de la página para abrir la página de configuración avanzada.

2. En la página de advanced settings, modifique los valores de los parámetros según sea necesario.

6. Resource group

Puede hacer clic en Configure Resource Group en la esquina superior derecha para consultar y cambiar el resource group actual de la tarea.

7. Ejecute la tarea de sincronización

1. Después de completar todos los ajustes, haga clic en Complete en la parte inferior de la página.

2. En la página Data Integration > Synchronization Task, busque la tarea que creó y haga clic en Start en la columna Operations.

3. Haga clic en el Name/ID de la tarea correspondiente en la Task List para consultar su proceso de ejecución detallado.

Consulte el estado de la tarea de sincronización

Una vez creada la tarea de sincronización, vaya a la página de tareas para consultar todas las tareas de sincronización del espacio de trabajo y su información básica.

La lista de tareas muestra las siguientes columnas: Name/ID, Data Source Synchronization Solution (tipo de sincronización, origen y destino), Status, Execution Overview, resource group y Actions.

• Puede Start o Stop una tarea de sincronización en la columna Actions. En More, puede realizar otras operaciones, como Edit y View.

• Para las tareas iniciadas, puede consultar el estado básico de ejecución en el Execution Overview y hacer clic en el área de resumen correspondiente para ver los detalles de ejecución.

Una tarea de sincronización en tiempo real desde una tabla de Hologres a Kafka consta de los siguientes tres pasos:

• Structure Migration: Incluye el método de creación de la tabla de destino (tabla existente o creación automática de tabla). Si selecciona la creación automática de tabla, se muestra la instrucción de lenguaje de definición de datos (DDL) para crear la tabla.

• Full initialization: Si selecciona Full Sync para el paso de sincronización de su tarea, el progreso de la inicialización completa se muestra aquí.

• Real-time Data Synchronization: Contiene estadísticas sobre la sincronización en tiempo real, incluido el tráfico de lectura/escritura en tiempo real, datos sucios, failovers y registros de operaciones.

Vuelva a ejecutar una tarea

En casos especiales, como cuando necesita modificar campos sincronizados o ajustar la información de la tabla de destino, puede hacer clic en Rerun en la columna Operations de la tarea de sincronización. Esta acción sincroniza los campos ajustados y otros cambios con el destino. El proceso omite las tablas no modificadas que ya se sincronizaron previamente.

• Para volver a ejecutar la tarea sin ningún cambio, haga clic en Rerun.

• Si edita la tarea, haga clic en Complete después de realizar los cambios. La acción de la tarea cambia a Apply Updates. Al hacer clic en Apply Updates, la tarea se vuelve a ejecutar con la nueva configuración.

Canal CDC

Canal CDC es un formato de datos CDC definido por Alibaba Canal.

• Campos y significados

  Nombre del campo	Significado del valor del campo
  id	El valor de este campo es fijo: 0.
  database	El nombre de la base de datos de Hologres.
  table	El nombre de la tabla de Hologres.
  pkNames	La columna de clave principal de la tabla de Hologres.
  isDdl	Indica si los registros binarios registran cambios DDL. El valor de este campo es fijo: false, ya que no se admite la sincronización de mensajes para cambios DDL en la tabla de Hologres.
  type	El tipo de cambio DML. Valores válidos: INSERT, UPDATE y DELETE. Nota Una operación de cambio en la tabla de Hologres genera dos registros cuyo tipo es UPDATE. Estos registros se escribirán en el topic de Kafka. Uno de los registros corresponde al contenido de los datos antes del cambio. El otro corresponde al contenido de los datos después del cambio. Si desea sincronizar datos completos desde la tabla de Hologres al topic de Kafka, el valor del campo type es fijo: INSERT.
  es	Una marca de tiempo de 13 dígitos en milisegundos. La marca de tiempo indica el momento en que se modificaron los datos en la tabla de Hologres. Si desea sincronizar datos completos desde la tabla de Hologres al topic de Kafka, el valor del campo es es fijo: 0.
  ts	Una marca de tiempo de 13 dígitos en milisegundos. La marca de tiempo indica el momento en que la tarea de sincronización lee los registros binarios generados para la tabla de Hologres.
  sql	El código SQL que registra los cambios DDL cuando los registros binarios generados para la tabla de Hologres contienen dichos cambios. El valor de este campo es fijo como una cadena vacía, ya que no se admite la sincronización de mensajes para cambios DDL en la tabla de Hologres.
  sqlType	Los tipos de datos de campo SQL a los que corresponden los tipos de datos de los campos de la tabla de Hologres. Asignaciones entre los tipos de datos de Hologres y los valores válidos de sqlType: bigint: -5 decimal cuya escala no es 0: 3 decimal cuya escala es 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	Los tipos de datos de campo MySQL a los que corresponden los tipos de datos de los campos de la tabla de Hologres. Asignaciones entre los tipos de datos de Hologres y los valores válidos de mysqlType: bigint: BIGINT int4: INT decimal cuya escala no es 0: DECIMAL(xx,xx) decimal cuya escala es 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	Los cambios de datos en la tabla de Hologres. Los nombres de campo en la tabla de Hologres se utilizan como claves, y los cambios de datos en los campos se serializan en cadenas y se utilizan como valores. Luego, las claves y los valores se organizan como cadenas con formato JSON. Para obtener más información sobre la serialización, consulte Description of JSON serialization.
  old	Se generan dos registros cuyo tipo es UPDATE para una operación de cambio en la tabla de Hologres y se escriben en el topic de Kafka. Los registros corresponden al contenido de los datos antes y después del cambio. En el primer registro, el campo old se utiliza para registrar el contenido de los datos en la tabla de Hologres antes del cambio. El campo data se utiliza para registrar el contenido de los datos para cambios DML distintos de UPDATE.

• Ejemplo de datos con formato Canal JSON que corresponden a cambios de datos generados por una operación INSERT en los registros binarios de la tabla de 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
  }

• Ejemplo de datos con formato Canal JSON que corresponden a datos completos sincronizados desde la tabla de 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
  }

• Ejemplo de dos registros de datos con formato Canal JSON que corresponden a cambios de datos generados por una operación UPDATE en los registros binarios de la tabla de 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
  }

• Ejemplo de datos con formato Canal JSON que corresponden a cambios de datos generados por una operación DELETE en los registros binarios de la tabla de 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

El formato JSON utiliza los nombres de los campos del registro binario de Hologres como claves y serializa los valores de los campos en cadenas. Los pares clave-valor resultantes se escriben en el topic de Kafka como cadenas con formato JSON.