Origen de datos de AnalyticDB for PostgreSQL - DataWorks - Centro de documentación de Alibaba Cloud

El origen de datos de AnalyticDB for PostgreSQL ofrece capacidades de lectura y escritura bidireccionales mediante Data Integration.

Limitaciones

La sincronización por lotes permite leer datos desde una vista.

Versiones compatibles

El conector es compatible con las versiones hasta la 7.0, incluida esta última.

Tipos de datos compatibles

Lectura por lotes

AnalyticDB for PostgreSQL Reader es compatible con la mayoría de los tipos de datos de AnalyticDB for PostgreSQL. Compruebe que sus tipos de datos sean compatibles antes de usarlos.

A continuación se muestran los tipos de datos compatibles para las lecturas por lotes.

Categoría	Tipo de AnalyticDB for PostgreSQL
Entero	BIGINT, BIGSERIAL, INTEGER, SMALLINT, SERIAL y GEOMETRY
Punto flotante	DOUBLE, PRECISION, MONEY, NUMERIC y REAL
Cadena	VARCHAR, CHAR, TEXT, BIT e INET
Fecha y hora	DATE, TIME y TIMESTAMP
Booleano	BOOL
Binario	BYTEA

Escritura por lotes

AnalyticDB for PostgreSQL Writer es compatible con la mayoría de los tipos de datos de AnalyticDB for PostgreSQL. Compruebe que sus tipos de datos sean compatibles antes de usarlos.

En la siguiente tabla se muestra la correspondencia entre los tipos de datos de DataWorks y los de AnalyticDB for PostgreSQL para las escrituras por lotes.

Tipo de datos de DataWorks	Tipo de AnalyticDB for PostgreSQL
LONG	BIGINT, BIGSERIAL, INTEGER, SMALLINT y SERIAL
DOUBLE	DOUBLE, PRECISION, MONEY, NUMERIC y REAL
STRING	VARCHAR, CHAR, TEXT, BIT, INET y GEOMETRY
DATE	DATE, TIME y TIMESTAMP
BOOLEAN	BOOL
BYTES	BYTEA

Nota

Debe usar una conversión de tipo, como a_inet::varchar, para los tipos de datos MONEY, INET y BIT.

Agregar un origen de datos

Antes de desarrollar una tarea de sincronización en DataWorks, agregue el origen de datos necesario a DataWorks siguiendo las instrucciones de Data source configuration. Consulte las descripciones de parámetros en la consola de DataWorks para comprender el significado de cada parámetro al agregar un origen de datos.

Desarrollar una tarea de sincronización de datos

Para conocer el punto de entrada y el procedimiento de configuración de una tarea de sincronización, consulte las siguientes guías de configuración.

Sincronización por lotes de una sola tabla

Para consultar el procedimiento, consulte Configure a sync task using the Codeless UI y Configure a sync task using the Code Editor.
Si necesita la lista completa de parámetros y ejemplos de scripts para Code Editor, revise Appendix: Script samples and parameters.

Sincronización por lotes de una base de datos completa

Para consultar el procedimiento, consulte Configure a batch synchronization task for an entire database.

Apéndice: Ejemplos de scripts y parámetros

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

Si desea configurar una tarea de sincronización por lotes con el editor de código, defina los parámetros correspondientes en el script según los requisitos unificados de formato. Para obtener más información, consulte Script mode configuration. A continuación se describen los parámetros que debe configurar para los orígenes de datos cuando utiliza el editor de código.

Ejemplo de script de Reader

{
    "type": "job",
    "steps": [
        {
            "parameter": {
                "datasource": "test_004",// The name of the data source.
                "column": [// The columns in the source table.
                    "id",
                    "name",
                    "sex",
                    "salary",
                    "age"
                ],
                "where": "id=1001",// The WHERE clause.
                "splitPk": "id",// The shard key.
                "table": "public.person"// The name of the source table.
            },
            "name": "Reader",
            "category": "reader"
        },
        {
            "parameter": {},
            "name": "Writer",
            "category": "writer"
        }
    ],
    "version": "2.0",// The version number.
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    },
    "setting": {
        "errorLimit": {// The error record count.
            "record": ""
        },
        "speed": {
            "concurrent": 6,// The number of concurrent threads.
            "throttle": true,// Specifies whether to enable throttling. If you set this parameter to false, throttling is disabled and the mbps parameter does not take effect. If you set this parameter to true, throttling is enabled.
           "mbps":"12"// The throttling rate in MB/s.
        }
    }
}

Parámetros de Reader

Parámetro	Descripción	Obligatorio	Predeterminado
datasource	Nombre del origen de datos registrado en DataWorks.	Sí	Ninguna
table	Nombre de la tabla de origen.	Sí	Ninguna
column	Columnas que se van a sincronizar, especificadas como un arreglo JSON. De forma predeterminada, se sincronizan todas las columnas, lo cual se indica con [*]. Seleccione un subconjunto de columnas para exportarlas. Cambie el orden de las columnas si lo requiere. Configure constantes respetando la sintaxis de SQL. Por ejemplo: `["id", "table","1","'mingya.wmy'","'null'", "to_char(a+1)","2.3","true"]`. id es un nombre de columna común. table es un nombre de columna que también funciona como palabra clave reservada. 1 es una constante entera. 'mingya.wmy' es una constante de cadena. El valor debe ir entre comillas simples ('). 'null' es una constante de cadena. to_char(a+1) es una expresión de función. 2.3 es un número de punto flotante. true es un valor booleano. Especifique explícitamente las columnas que se van a sincronizar en el parámetro column. Este valor no puede quedar vacío.	Sí	Ninguna
splitPk	Clave de fragmentación. Establezca splitPk con el nombre de una columna para habilitar la extracción paralela de datos mediante fragmentación. Se recomienda usar la clave principal de la tabla, ya que suele distribuirse de manera uniforme y ayuda a evitar puntos críticos de datos. El parámetro splitPk solo admite la fragmentación de datos en columnas de tipo entero. Si especifica una columna de otro tipo de datos, como cadena, flotante o fecha, Data Integration omitirá la configuración de splitPk y sincronizará los datos en un solo hilo. Si no configura el parámetro splitPk o lo deja vacío, la tarea sincronizará los datos con un único hilo.	No	Ninguna
where	Cláusula WHERE que se usa para filtrar los datos extraídos. El lector construye una consulta SQL a partir de los parámetros column, table y where. Por ejemplo: `id>2 and sex=1`. Utilice la cláusula WHERE para ejecutar una sincronización incremental. Si no especifica ninguna cláusula WHERE, el conector sincronizará toda la tabla.	No	Ninguna
querySql (avanzado; no disponible en Codeless UI)	Consulta SQL personalizada para filtrar datos. Cuando el parámetro `where` no alcanza para expresar sus condiciones de filtrado, utilice este parámetro en su lugar. Data Integration ignorará los parámetros column, table y where si configura querySql. Por ejemplo: `select a,b from table_a join table_b on table_a.id = table_b.id`. Cuando configura el parámetro querySql, AnalyticDB for PostgreSQL Reader ignora los parámetros column, table y where.	No	Ninguna
fetchSize	Número de registros que se obtienen por lote desde la base de datos. Un valor mayor reduce los viajes de ida y vuelta por la red y mejora el rendimiento de la extracción. Nota Un valor de fetchSize superior a 2048 puede provocar un error de memoria insuficiente (OOM) durante el proceso de sincronización de datos.	No	512

Ejemplo de script de Writer

{
    "type": "job",
    "steps": [
        {
            "parameter": {},
            "name": "Reader",
            "category": "reader"
        },
        {
            "parameter": {
                "postSql": [],// The post-execution SQL statement.
                "datasource": "test_004",// The name of the data source.
                "column": [// The columns in the destination table.
                    "id",
                    "name",
                    "sex",
                    "salary",
                    "age"
                ],
                "table": "public.person",// The name of the destination table.
                "preSql": []// The pre-execution SQL statement.
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "version": "2.0",// The version number.
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    },
    "setting": {
        "errorLimit": {// The error record count.
            "record": ""
        },
        "speed": {
            "throttle":true,// Specifies whether to enable throttling. If you set this parameter to false, throttling is disabled and the mbps parameter does not take effect. If you set this parameter to true, throttling is enabled.
            "concurrent":6, // The number of concurrent threads.
            "mbps":"12"// The throttling rate.
        }
    }
}

Parámetros de Writer

Parámetro	Descripción	Obligatorio	Predeterminado
datasource	Nombre del origen de datos registrado en DataWorks.	Sí	Ninguna
table	Nombre de la tabla de destino.	Sí	Ninguna
writeMode	Modo de escritura. Valores válidos: insert, copy y upsert. insert: ejecuta la instrucción de PostgreSQL `INSERT INTO ... VALUES ...` para escribir datos en PostgreSQL. Se recomienda este modo. copy: PostgreSQL proporciona el comando COPY para copiar datos entre una tabla y un archivo (entrada estándar o salida estándar). Data Integration admite el uso de `COPY FROM` para cargar datos en una tabla. Utilice este modo para mejorar el rendimiento. upsert: si ocurre un conflicto de datos durante la operación de escritura, el sistema gestiona los datos nuevos y los existentes según el parámetro conflictMode.	No	insert
conflictMode	Si establece writeMode en upsert y se produce un conflicto de clave principal o de índice único al escribir datos en PostgreSQL, seleccione una de las siguientes políticas de resolución de conflictos: replace: si ocurre un conflicto, los datos entrantes sobrescriben los existentes. ignore: ante un conflicto, el conector descarta los datos nuevos y conserva los existentes. Nota La política de resolución de conflictos solo se configura en Code Editor.	No	replace
column	Columnas de destino donde se escribirán los datos, separadas por comas (,). Por ejemplo: `"column":["id","name","age"]`. Para escribir en todas las columnas en orden, use un asterisco (). Por ejemplo: "column":[""].	Sí	Ninguna
preSql	Instrucción SQL que se ejecuta antes de iniciar la tarea de sincronización. Codeless UI admite una sola instrucción, mientras que Code Editor permite varias, por ejemplo, para limpiar datos antiguos.	No	Ninguna
postSql	Instrucción SQL que se ejecuta después de completar la tarea de sincronización. Codeless UI admite una sola instrucción, pero Code Editor permite varias, por ejemplo, para agregar una marca de tiempo.	No	Ninguna
batchSize	Número de registros que se escriben por lote. Un valor mayor reduce los viajes de ida y vuelta por la red y mejora el rendimiento, aunque un valor excesivo puede provocar un error OOM.	No	1024