Realtime Compute for Apache Flink:Acceso a AnalyticDB for PostgreSQL

Limitaciones

• Realtime Compute for Apache Flink no admite la lectura desde AnalyticDB for PostgreSQL en modo serverless.

• El conector de AnalyticDB for PostgreSQL requiere Ververica Runtime (VVR) 6.0.0 o posterior.

• AnalyticDB for PostgreSQL V7.0 requiere VVR 8.0.1 o posterior.

Si desea utilizar un conector personalizado, consulte Manage custom connectors .

Requisitos previos

Antes de empezar, asegúrese de contar con lo siguiente:

• Un espacio de trabajo de Flink totalmente administrado. Consulte Activate fully managed Flink.

• Una instancia de AnalyticDB for PostgreSQL y una cuenta con privilegios. Consulte Create an instance y Create a privileged account.

• La instancia de AnalyticDB for PostgreSQL y el espacio de trabajo de Flink totalmente administrado deben estar en la misma virtual private cloud (VPC).

Si se encuentran en distintas VPC, consulte How does fully managed Flink access a service across VPCs?

Paso 1: Configurar una lista de permitidos y preparar los datos

1. Inicie sesión en AnalyticDB for PostgreSQL console.

2. Agregue el bloque CIDR del espacio de trabajo de Flink totalmente administrado a la lista de permitidos de la instancia de AnalyticDB for PostgreSQL.

   1. Localice el bloque CIDR del vSwitch que utiliza su espacio de trabajo de Flink totalmente administrado. Consulte How do I configure a whitelist?

   2. Agregue el bloque CIDR a la lista de permitidos de la instancia de AnalyticDB for PostgreSQL. Consulte Procedure.

   Si accede a la instancia a través de Internet, agregue la dirección IP pública en su lugar.

3. En la página de detalles de la instancia, haga clic en Log On to Database en la esquina superior derecha e introduzca su nombre de usuario y contraseña. Consulte Use client tools to connect to an instance para obtener más información.

4. Cree una tabla de dimensiones llamada adbpg_dim_table e inserte 50 filas de datos de ejemplo.

   -- Create the dimension table
   CREATE TABLE adbpg_dim_table(
     id int,
     username text,
     PRIMARY KEY(id)
   );
   
   -- Insert 50 rows: id ranges from 1 to 50, username is "username" followed by the row number
   INSERT INTO adbpg_dim_table(id, username)
   SELECT i, 'username'||i::text
   FROM generate_series(1, 50) AS t(i);

   Ejecute SELECT * FROM adbpg_dim_table ORDER BY id; para comprobar los datos insertados.

5. Cree una tabla de resultados llamada adbpg_sink_table donde Flink escribirá la salida.

   CREATE TABLE adbpg_sink_table(
     id int,
     username text,
     score int
   );

Paso 2: Crear un borrador de flujo

1. Inicie sesión en Realtime Compute for Apache Flink console, localice su espacio de trabajo y haga clic en Console en la columna Actions.

2. En el panel de navegación izquierdo, vaya a Development > ETL. En la esquina superior izquierda de la página del editor SQL, haga clic en New y seleccione New Blank Stream Draft.

3. En el cuadro de diálogo New Draft, configure los siguientes parámetros.

   Parámetro	Descripción	Ejemplo
   Name	Nombre del borrador. Debe ser único dentro del proyecto.	adbpg-test
   Location	Carpeta donde se guarda el borrador. Haga clic en el icono junto a una carpeta existente para crear una subcarpeta.	Draft
   Engine Version	Versión del motor Flink. Consulte Engine versions para conocer los detalles de las versiones y su ciclo de vida.	vvr-8.0.1-flink-1.17

4. Haga clic en Create.

Paso 3: Escribir y desplegar el borrador

1. Copie el siguiente código SQL en el editor. Este SQL define tres tablas y una unión de búsqueda que enriquece el flujo de Datagen con datos de usuarios provenientes de AnalyticDB for PostgreSQL.

   -- Source table: Datagen generates sequential IDs (1-50) and random scores (70-100).
   -- No changes needed in the WITH clause for this example.
   CREATE TEMPORARY TABLE datagen_source (
     id INT,
     score INT
   ) WITH (
     'connector' = 'datagen',
     'fields.id.kind' = 'sequence',
     'fields.id.start' = '1',
     'fields.id.end' = '50',
     'fields.score.kind' = 'random',
     'fields.score.min' = '70',
     'fields.score.max' = '100'
   );
   
   -- Dimension table: backed by AnalyticDB for PostgreSQL.
   -- Flink queries this table at processing time to look up usernames by ID.
   -- Replace the WITH clause values with your actual connection details.
   CREATE TEMPORARY TABLE dim_adbpg(
     id int,
     username varchar,
     PRIMARY KEY(id) NOT ENFORCED
   ) WITH (
     'connector' = 'adbpg',
     'url' = 'jdbc:postgresql://gp-2ze****3tysk255b5-master.gpdb.rds.aliyuncs.com:5432/flinktest',
     'tablename' = 'adbpg_dim_table',
     'username' = 'flinktest',
     'password' = '${secret_values.adb_password}',
     'maxRetryTimes' = '2',
     'cache' = 'lru',
     'cacheSize' = '100'
   );
   
   -- Result table: Flink writes enriched records here.
   -- Replace the WITH clause values with your actual connection details.
   CREATE TEMPORARY TABLE sink_adbpg (
     id int,
     username varchar,
     score int
   ) WITH (
     'connector' = 'adbpg',
     'url' = 'jdbc:postgresql://gp-2ze****3tysk255b5-master.gpdb.rds.aliyuncs.com:5432/flinktest',
     'tablename' = 'adbpg_sink_table',
     'username' = 'flinktest',
     'password' = '${secret_values.adb_password}',
     'maxRetryTimes' = '2',
     'conflictMode' = 'ignore',
     'retryWaitTime' = '200'
   );
   
   -- Lookup join: for each record from datagen_source, Flink looks up the matching
   -- row in dim_adbpg at the time the record is processed (PROCTIME()).
   INSERT INTO sink_adbpg
   SELECT ts.id, ts.username, ds.score
   FROM datagen_source AS ds
   JOIN dim_adbpg FOR SYSTEM_TIME AS OF PROCTIME() AS ts
   ON ds.id = ts.id;

   Sobre la sintaxis de la unión de búsqueda: FOR SYSTEM_TIME AS OF PROCTIME() indica a Flink que debe consultar la tabla de dimensiones en el momento exacto en que se procesa cada registro de origen. Esto significa que cada registro se enriquece con los datos de dimensiones vigentes al momento del procesamiento, por lo que los resultados ya escritos no se actualizarán si la tabla de dimensiones cambia posteriormente.

2. Actualice los parámetros de conexión de la tabla de dimensiones y la tabla de resultados. Reemplace los valores de marcador de posición en las cláusulas WITH con sus detalles reales de conexión a AnalyticDB for PostgreSQL. La tabla de origen Datagen no requiere cambios. Para consultar la referencia completa de parámetros y las asignaciones de tipos de datos, consulte AnalyticDB for PostgreSQL connector.

   Parámetro	Obligatorio	Predeterminado	Descripción
   url	Sí	—	URL JDBC con el formato jdbc:postgresql://<Internal endpoint>:<Port>/<Database name>. Puede encontrarla en la página Database Connection de la instancia dentro de la consola de AnalyticDB for PostgreSQL.
   tablename	Sí	—	Nombre de la tabla en la base de datos de AnalyticDB for PostgreSQL.
   username	Sí	—	Nombre de usuario para acceder a la base de datos.
   password	Sí	—	Contraseña de la cuenta de la base de datos.
   targetSchema	No	public	Nombre del esquema. Especifíquelo únicamente si su tabla no se encuentra en el esquema public.
   maxRetryTimes	No	—	Número máximo de reintentos tras un fallo de escritura.
   cache	No	—	Directiva de caché para las búsquedas en la tabla de dimensiones. Establezca este valor en lru para mantener en memoria las entradas consultadas recientemente. El almacenamiento en caché LRU reduce el tráfico hacia la base de datos y mejora el throughput de las búsquedas, aunque las entradas almacenadas podrían quedar desactualizadas. Se trata de un equilibrio entre throughput y actualidad de los datos; ajuste el valor de cacheSize y evalúe su tolerancia a datos obsoletos antes de habilitar esta opción.
   cacheSize	No	—	Número máximo de entradas que se almacenan en caché. Los valores más altos reducen las solicitudes a la base de datos, pero consumen más memoria.
   conflictMode	No	—	Acción que se ejecuta cuando una escritura entra en conflicto con una clave primaria o un índice existente. Establezca este valor en ignore para omitir las filas conflictivas.
   retryWaitTime	No	—	Tiempo de espera en milisegundos entre los reintentos de escritura.

3. En la esquina superior derecha de la página del editor SQL, haga clic en Validate para comprobar la sintaxis.

4. Haga clic en Deploy.

5. En la página O&M > Deployments, localice su despliegue y haga clic en Start en la columna Actions.

Paso 4: Comprobar el resultado

1. Inicie sesión en la consola de AnalyticDB for PostgreSQL.

2. Haga clic en Log On to Database. Consulte Connect to an instance from a client para obtener más información.

3. Ejecute la siguiente consulta para ver los registros que Flink escribió en la tabla de resultados.

   SELECT * FROM adbpg_sink_table ORDER BY id;

   El resultado debería contener 50 filas, cada una con un ID de usuario, el nombre de usuario correspondiente de la tabla de dimensiones y una puntuación aleatoria entre 70 y 100.

   La consulta devuelve 7 registros con tres columnas: id, username y score. Las filas 1 a 7 corresponden a username1 hasta username7, con puntuaciones de 94, 79, 70, 93, 71, 82 y 87 respectivamente.

Referencias

• Datagen connector

• AnalyticDB for PostgreSQL connector