Write data to AnalyticDB for PostgreSQL - AnalyticDB - Centro de documentación de Alibaba Cloud

Este tema describe cómo escribir datos en AnalyticDB for PostgreSQL mediante Realtime Compute for Apache Flink.

Límites de uso

Esta función no es compatible con el modo Serverless de AnalyticDB for PostgreSQL.
El conector de AnalyticDB for PostgreSQL solo es compatible con Realtime Compute for Apache Flink VVR 6.0.0 o versiones posteriores.
AnalyticDB for PostgreSQL V7.0 solo es compatible con Realtime Compute for Apache Flink VVR 8.0.1 o versiones posteriores.

Nota
Si utiliza un conector personalizado, consulte Manage custom connectors para obtener más información.

Requisitos previos

Se ha creado un espacio de trabajo de Fully Managed Flink. Para obtener más información, consulte Activate Realtime Compute for Apache Flink.
Se ha creado una instancia de AnalyticDB for PostgreSQL. Para obtener más información, consulte Create an instance.
La instancia de AnalyticDB for PostgreSQL y el espacio de trabajo de Fully Managed Flink deben desplegarse en la misma VPC.

Configuración de una instancia de AnalyticDB for PostgreSQL

Inicie sesión en la consola de AnalyticDB for PostgreSQL.
Agregue el bloque CIDR del espacio de trabajo de Flink a la lista de permitidos de la instancia de AnalyticDB for PostgreSQL. Para obtener más información sobre cómo configurar una lista de permitidos, consulte Configure a whitelist.
Haga clic en Log On to Database. Para obtener más información sobre otras formas de conectarse a la base de datos, consulte Connect to an instance by using a client.
Cree una tabla en la instancia de AnalyticDB for PostgreSQL.
Sentencia SQL de ejemplo para crear una tabla:
```
CREATE TABLE test_adbpg_table(
b1 int,
b2 int,
b3 text,
PRIMARY KEY(b1)
);
```

Configuración de Realtime Compute for Flink

Inicie sesión en la consola de Realtime Compute.
En la pestaña Fully Managed Flink, haga clic en Console en la columna Actions del espacio de trabajo de destino.
En el panel de navegación izquierdo, haga clic en Connectors.
En la página Connectors, haga clic en Create Custom Connector.
Cargue el archivo JAR del conector personalizado.
Nota
- Obtenga el paquete JAR del conector personalizado de AnalyticDB for PostgreSQL para Flink. Para obtener más información, consulte AnalyticDB PostgreSQL Connector.
- La versión del paquete JAR debe coincidir con la versión del motor de Flink de la plataforma Realtime Compute.
Una vez completada la carga, haga clic en Next.

El sistema analiza sintácticamente el conector personalizado cargado. Si el análisis es exitoso, puede continuar con el siguiente paso. Si falla, compruebe si el código del conector personalizado cumple con los estándares de la comunidad de Apache Flink.
Haga clic en Finish.

El conector personalizado creado aparece en la lista de conectores.

Creación de un trabajo de Flink

Inicie sesión en la consola de Realtime Compute. En la pestaña Fully Managed Flink, haga clic en Console en la columna Actions del espacio de trabajo de destino.
En el panel de navegación izquierdo, haga clic en SQL Development. Haga clic en New, seleccione Blank Streaming Job Draft y luego haga clic en Next.

En el cuadro de diálogo New File Draft, configure los parámetros del trabajo.

Parámetro del trabajo	Descripción	Ejemplo
Name	El nombre del trabajo. Nota El nombre del trabajo debe ser único dentro del proyecto actual.	adbpg-test
Location	La carpeta a la que pertenece el archivo de código del trabajo. También puede hacer clic en el icono a la derecha de una carpeta existente para crear una subcarpeta.	Job Drafts
Engine Version	La versión del motor de Flink que utiliza el trabajo actual. Para obtener más información sobre los números de versión del motor, las asignaciones de versiones y los hitos del ciclo de vida, consulte Engine versions.	vvr-6.0.7-flink-1.15

Haga clic en Create.

Escritura de datos en AnalyticDB for PostgreSQL

Escriba el código del trabajo.

Cree una tabla de origen aleatoria datagen_source y una tabla de destino test_adbpg_table en AnalyticDB for PostgreSQL. Copie el siguiente código de trabajo en el editor de texto del trabajo.

CREATE TABLE datagen_source (
 f_sequence INT,
 f_random INT,
 f_random_str STRING
) WITH (
 'connector' = 'datagen',
 'rows-per-second'='5',
 'fields.f_sequence.kind'='sequence',
 'fields.f_sequence.start'='1',
 'fields.f_sequence.end'='1000',
 'fields.f_random.min'='1',
 'fields.f_random.max'='1000',
 'fields.f_random_str.length'='10'
);
CREATE TABLE test_adbpg_table (
    `B1` bigint   ,
    `B2` bigint  ,
    `B3` VARCHAR ,
    `B4` VARCHAR,
     PRIMARY KEY(B1) not ENFORCED
) with (
   'connector' = 'adbpg-nightly-1.13',
   'password' = 'xxx',
   'tablename' = 'test_adbpg_table',
   'username' = 'xxxx',
   'url' = 'jdbc:postgresql://url:5432/schema',
   'maxretrytimes' = '2',
   'batchsize' = '50000',
   'connectionmaxactive' = '5',
   'conflictmode' = 'ignore',
   'usecopy' = '0',
   'targetschema' = 'public',
   'exceptionmode' = 'ignore',
   'casesensitive' = '0',
   'writemode' = '1',
   'retrywaittime' = '200'
);

No es necesario modificar los parámetros de la tabla datagen_source. Debe modificar los parámetros de la tabla test_adbpg_table según sus requisitos comerciales reales. La siguiente tabla describe los parámetros.

Parámetro	Obligatorio	Descripción
connector	Sí	El nombre del conector. Establezca este parámetro en `adbpg-nightly-<número de versión>`, por ejemplo, `adbpg-nightly-1.13`.
url	Sí	La URL JDBC de la instancia de AnalyticDB for PostgreSQL. Formato: `jdbc:postgresql://<endpoint interno>:<puerto>/<nombre de la base de datos>`. Ejemplo: `jdbc:postgresql://gp-xxxxxx.gpdb.cn-chengdu.rds.aliyuncs.com:5432/postgres`.
tablename	Sí	El nombre de la tabla de AnalyticDB for PostgreSQL.
username	Sí	La cuenta de la base de datos de la instancia de AnalyticDB for PostgreSQL.
password	Sí	La contraseña de la cuenta de la base de datos para la instancia de AnalyticDB for PostgreSQL.
maxretrytimes	No	El número máximo de reintentos tras un fallo en la ejecución de SQL. Valor predeterminado: 3.
batchsize	No	El número máximo de registros de datos que se escriben en un solo lote. Valor predeterminado: 50000.
exceptionmode	No	La directiva de manejo de errores cuando ocurre una excepción durante la escritura de datos. Valores válidos: ignore: omite los datos que causan una excepción. Este es el valor predeterminado. strict: desencadena un failover y reporta un error cuando ocurre una excepción de escritura de datos.
conflictmode	No	La directiva para manejar conflictos de clave primaria o índice único. Valores válidos: ignore: omite los conflictos de clave primaria y conserva los datos existentes. strict: desencadena un failover y reporta un error cuando ocurre un conflicto de clave primaria. update: actualiza los datos cuando ocurre un conflicto de clave primaria. upsert: escribe datos mediante el método UPSERT cuando ocurre un conflicto de clave primaria. Este es el valor predeterminado. AnalyticDB for PostgreSQL implementa UPSERT mediante INSERT ON CONFLICT y COPY ON CONFLICT. Si la tabla de destino está particionada, la versión menor del kernel debe ser V6.3.6.1 o posterior. Para obtener más información sobre cómo actualizar la versión menor del kernel, consulte Upgrade the engine version.
targetschema	No	El esquema de la base de datos de AnalyticDB for PostgreSQL. Valor predeterminado: public.
writemode	No	El modo de escritura de datos. Valores válidos: 0: escribe datos mediante BATCH INSERT. 1: escribe datos mediante la API COPY. Este es el valor predeterminado. 2: escribe datos mediante BATCH UPSERT.
verbose	No	Especifica si se muestran los registros de tiempo de ejecución del conector. Valores válidos: 0: no muestra los registros de tiempo de ejecución. Este es el valor predeterminado. 1: muestra los registros de tiempo de ejecución.
retrywaittime	No	El intervalo entre intentos de reintento cuando ocurre una excepción. Unidad: milisegundos. Valor predeterminado: 100.
batchwritetimeoutms	No	El tiempo máximo de acumulación por lotes para la escritura en lotes. Cuando se supera este tiempo, se escribe el lote acumulado. Unidad: milisegundos. Valor predeterminado: 50000.
connectionmaxactive	No	El parámetro del pool de conexiones. Este parámetro especifica el número máximo de conexiones simultáneas en el pool de conexiones de un solo Task Manager. Valor predeterminado: 5.
casesensitive	No	Especifica si los nombres de columnas y tablas distinguen entre mayúsculas y minúsculas. Valores válidos: 0: no distingue entre mayúsculas y minúsculas. Este es el valor predeterminado. 1: distingue entre mayúsculas y minúsculas.

Nota

Se admiten asignaciones de parámetros y tipos. Para obtener más información, consulte la documentación del conector en AnalyticDB for PostgreSQL (ADB PG).

Inicie el trabajo.
1. En la parte superior de la página de desarrollo de trabajos, haga clic en Deploy. En el cuadro de diálogo que aparece, haga clic en OK.
  
  Nota
  Los clústeres de sesión son adecuados para desarrollo y pruebas en entornos que no son de producción. Puede usar clústeres de sesión para depurar trabajos, mejorar la utilización de recursos del Job Manager (JM) y acelerar el inicio de los trabajos. Sin embargo, no se recomienda enviar trabajos a clústeres de sesión debido a preocupaciones sobre la estabilidad del negocio. Para obtener más información, consulte Debug a job.
2. En la página Deployments, haga clic en Resume en la columna Actions del trabajo de destino.
3. Haga clic en Resume.

Verificación de los resultados

Conéctese a la base de datos de AnalyticDB for PostgreSQL. Para obtener más información, consulte Connect to an instance by using a client.
Ejecute la siguiente sentencia para consultar la tabla test_adbpg_table.
```
SELECT * FROM test_adbpg_table;
```
Los datos se escriben en AnalyticDB for PostgreSQL como se esperaba. La siguiente figura muestra un retorno de ejemplo.

El resultado de la consulta contiene tres columnas: b1 (int4), b2 (int4) y b3 (text). Se devuelven varias filas de datos, lo que confirma que los datos se han sincronizado con la tabla de destino.

AnalyticDB:Escritura de datos en AnalyticDB for PostgreSQL