DataWorks:Origen de datos PolarDB

Lectura y escritura por lotes

Es posible leer datos desde vistas.

Lectura en tiempo real

Si utiliza un clúster de PolarDB for MySQL como origen, debe habilitar el registro binario. PolarDB for MySQL es una base de datos nativa de la nube totalmente compatible con MySQL. De forma predeterminada, PolarDB for MySQL utiliza registros físicos de alto nivel en lugar del registro binario. Para una mejor integración con el ecosistema de MySQL, PolarDB permite habilitar el registro binario.

Lectura por lotes

La siguiente tabla muestra las asignaciones de tipos de datos para PolarDB Reader.

Escritura por lotes

Al igual que PolarDB Reader, PolarDB Writer admite la mayoría de los tipos de datos de PolarDB, pero no todos. Asegúrese de que sus tipos de datos sean compatibles.

La siguiente tabla muestra las asignaciones de tipos de datos para PolarDB Writer.

Configurar una lista de permitidos de direcciones IP

Agregue el bloque CIDR de la VPC que contiene su Serverless resource group o Exclusive Resource Group for Data Integration a la lista de permitidos de direcciones IP de su clúster de PolarDB. Para obtener más información, consulte Set a cluster whitelist.

Crear una cuenta y otorgar permisos

Cree una cuenta y otórguele los permisos necesarios.

Cree una cuenta de base de datos para la sincronización de datos. La cuenta debe tener los permisos SELECT, REPLICATION SLAVE, REPLICATION CLIENT sobre la base de datos.

1. Cree una cuenta.

   Para obtener más información, consulte Create and manage a database account.

2. Otorgue permisos.

   Ejecute el siguiente comando para otorgar los permisos necesarios a la cuenta. Alternativamente, puede otorgar el permiso SUPER.

   -- CREATE USER 'sync_account'@'%' IDENTIFIED BY 'your_password';
   GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'sync_account'@'%';

Habilitar el registro binario

Para obtener más información, consulte Enable binary logging.

Configurar una tarea por lotes de tabla única

• Para obtener más información, consulte Configure a task in the codeless UI y Configure a task in script mode.

• Para ver la lista completa de parámetros y una demostración de script, consulte Appendix: Script demo and parameters.

Configurar la sincronización en tiempo real para una tabla o base de datos

Para obtener más información, consulte Configure a real-time synchronization task (legacy).

Configurar la lectura por lotes de base de datos completa y la sincronización incremental en tiempo real

Para obtener más información, consulte Configure a real-time full-database synchronization task.

Configurar una tarea de sincronización por lotes mediante el editor de código

Si desea configurar una tarea de sincronización por lotes mediante el editor de código, debe configurar los parámetros relacionados en el script según los requisitos unificados de formato de script. Para obtener más información, consulte Script mode configuration. La siguiente información describe los parámetros que debe configurar para los orígenes de datos cuando configura una tarea de sincronización por lotes mediante el editor de código.

Demostración de script de Reader

El siguiente código proporciona un ejemplo de script para leer datos de una sola tabla en una base de datos. Para obtener más información sobre los parámetros, consulte la sección de descripción de parámetros.

{
    "type": "job",
    "steps": [
        {
            "parameter": {
                "datasource": "test_005",                // The name of the data source.
                "column": [                              // The source column names.
                    "id",
                    "name",
                    "age",
                    "sex",
                    "salary",
                    "interest"
                ],
                "where": "id=1001",                      // The filter condition.
                "splitPk": "id",                         // The sharding key.
                "table": "PolarDB_person",               // The source table name.
              	"useReadonly": "false"                   // Specifies whether to read data from a secondary database.
            },
            "name": "Reader",
            "category": "reader"
        },
        {
            "parameter": {}
    ],
    "version": "2.0",                                // The version number.
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    },
    "setting": {
        "errorLimit": {                              // The error tolerance.
            "record": ""
        },
        "speed": {
            "concurrent": 6,                         // The concurrency level.
            "throttle": true,                        // If throttle is set to false, the mbps parameter does not take effect and throttling is disabled. If throttle is set to true, throttling is enabled.
      "mbps":"12"                              // The throttling rate, in MB/s.
        }
    }
}

Parámetros del script de Reader

Demostración de script de Writer

El siguiente código proporciona una configuración de script de ejemplo. Para obtener más información sobre los parámetros, consulte la sección de descripción de parámetros.

{
    "type": "job",
    "steps": [
        {
            "parameter": {},
            "name": "Reader",
            "category": "reader"
        },
        {
            "parameter": {
                "postSql": [],                        // The SQL statement to be executed after the synchronization task is complete.
                "datasource": "test_005",             // The name of the data source.
                "column": [                           // The destination column names.
                    "id",
                    "name",
                    "age",
                    "sex",
                    "salary",
                    "interest"
                ],
                "writeMode": "insert",                // The write mode.
                "batchSize": 256,                     // The number of records to submit in each batch.
                "table": "PolarDB_person_copy",       // The destination table name.
                "preSql": []                          // The SQL statement to be executed before the synchronization task starts.
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "version": "2.0",                             // The version number.
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    },
    "setting": {
        "errorLimit": {                           // The error tolerance.
            "record": ""
        },
        "speed": {
            "throttle":true,                      // If throttle is set to false, the mbps parameter does not take effect and throttling is disabled. If throttle is set to true, throttling is enabled.
            "concurrent":6,                       // The concurrency level.
            "mbps":"12"                           // The throttling rate, in MB/s.
        }
    }
}

Parámetros del script de Writer

• Todos los parámetros

  Parámetro	Descripción	Obligatorio	Predeterminado
  datasource	El nombre del origen de datos. En modo script, el valor de este parámetro debe coincidir con el nombre del origen de datos agregado.	Sí	Ninguna
  table	El nombre de la tabla de destino en la cual desea sincronizar datos.	Sí	Ninguna
  writeMode	El modo de escritura. Valores válidos: insert: corresponde a INSERT INTO en la interfaz sin código. update: corresponde a ON DUPLICATE KEY UPDATE en la interfaz sin código. replace: corresponde a REPLACE INTO en la interfaz sin código. Para obtener más información sobre los modos y ejemplos, consulte la sección Detalles del parámetro writeMode a continuación. Nota Para PolarDB for PostgreSQL, solo se admite el modo insert. Para actualizar datos y evitar conflictos de clave principal, elimine los datos duplicados antes de ejecutar la tarea de sincronización por lotes. Se recomiendan los siguientes métodos: Método 1: En el parámetro preSql (que corresponde a Pre-Import Statement en la interfaz sin código), configure una sentencia TRUNCATE para borrar la tabla de destino. Método 2: Procese la tabla de destino en un nodo ascendente para evitar conflictos de clave principal durante la sincronización de datos.	No	insert
  column	Las columnas de destino en las que desea escribir datos. Separe las columnas con comas (,). Ejemplo: "column": ["id", "name", "age"]. Para escribir datos en todas las columnas en orden, utilice un asterisco (). Ejemplo: "column": [""].	Sí	Ninguna
  preSql	Especifica una o más sentencias SQL para ejecutar antes de que comience la tarea. La interfaz sin código admite una sola sentencia, mientras que el modo script admite múltiples sentencias, como sentencias para borrar datos existentes.	No	Ninguna
  postSql	Especifica una o más sentencias SQL para ejecutar después de que finalice la tarea. La interfaz sin código admite una sola sentencia, mientras que el modo script admite múltiples sentencias, como una sentencia para agregar una marca de tiempo.	No	Ninguna
  batchSize	El número de registros que se enviarán en cada lote. Valores más grandes pueden mejorar el throughput al reducir las interacciones de red con PolarDB, pero un valor excesivamente grande puede provocar un error de falta de memoria (OOM).	No	1024
  updateColumn	Las columnas que se actualizarán cuando ocurra un conflicto de clave principal o índice único. Este parámetro solo es válido cuando writeMode se establece en update. Puede especificar varias columnas separadas por comas. Ejemplo: "updateColumn": ["name", "age"]. Nota Este parámetro solo es compatible con PolarDB for MySQL.	No	Ninguna

• Detalles del parámetro writeMode

  Comparación	insert (corresponde a INSERT INTO en la interfaz sin código)	update (corresponde a ON DUPLICATE KEY UPDATE en la interfaz sin código)	replace (corresponde a REPLACE INTO en la interfaz sin código)
  Estrategia de manejo de conflictos	Si ocurre un conflicto de clave principal o índice único, la fila en conflicto no se escribe en la tabla de destino y se procesa como datos sucios.	Si no ocurre ningún conflicto de clave principal o índice único, este modo funciona igual que el modo INSERT INTO. Si ocurre un conflicto, la nueva fila actualiza solo los campos especificados de la fila existente.	Si no ocurre ningún conflicto de clave principal o índice único, este modo funciona igual que el modo INSERT INTO. Si ocurre un conflicto, la fila existente se elimina y se inserta la nueva fila, reemplazando efectivamente todos sus campos.
  Ejemplo de datos	Tabla de origen +----+---------+-----+ | id | name | age | +----+---------+-----+ | 1 | zhangsan| 1 | | 2 | lisi | | +----+---------+-----+ Tabla de destino original +----+---------+-----+ | id | name | age | +----+---------+-----+ | 2 | wangwu | | +----+---------+-----+ Después de ejecutar la tarea, se escribe una fila en la tabla de destino y se registra una fila como datos sucios. +----+---------+-----+ | id | name | age | +----+---------+-----+ | 1 | zhangsan| 1 | | 2 | wangwu | | +----+---------+-----+	Escenario 1: La tarea está configurada para sincronizar solo algunas columnas: "column": ["id","name"] Tabla de origen +----+---------+-----+ | id | name | age | +----+---------+-----+ | 1 | zhangsan| 1 | | 2 | lisi | | +----+---------+-----+ Tabla de destino original +----+---------+-----+ | id | name | age | +----+---------+-----+ | 2 | wangwu | 3 | +----+---------+-----+ Después de ejecutar la tarea, se escriben dos filas en la tabla de destino y no se registran datos sucios. +----+---------+-----+ | id | name | age | +----+---------+-----+ | 1 | zhangsan| 1 | | 2 | lisi | 3 | +----+---------+-----+ Escenario 2: La tarea está configurada para sincronizar todas las columnas: "column": ["id","name","age"] Tabla de origen +----+---------+-----+ | id | name | age | +----+---------+-----+ | 1 | zhangsan| 1 | | 2 | lisi | | +----+---------+-----+ Tabla de destino original +----+---------+-----+ | id | name | age | +----+---------+-----+ | 2 | wangwu | 3 | +----+---------+-----+ Después de ejecutar la tarea, se escriben dos filas en la tabla de destino y no se registran datos sucios. +----+---------+-----+ | id | name | age | +----+---------+-----+ | 1 | zhangsan| 1 | | 2 | lisi | | +----+---------+-----+	Tabla de origen +----+---------+-----+ | id | name | age | +----+---------+-----+ | 1 | zhangsan| 1 | | 2 | lisi | | +----+---------+-----+ Tabla de destino original +----+---------+-----+ | id | name | age | +----+---------+-----+ | 2 | wangwu | 3 | +----+---------+-----+ Después de ejecutar la tarea, se escriben dos filas en la tabla de destino y no se registran datos sucios. +----+---------+-----+ | id | name | age | +----+---------+-----+ | 1 | zhangsan| 1 | | 2 | lisi | | +----+---------+-----+