Batch import data into Hologres using Flink - Hologres - Centro de documentación de Alibaba Cloud

Importe datos por lotes a Hologres mediante el conector de Flink para lograr una ingesta de datos altamente eficiente y con baja carga.

Contexto

Hologres se integra con Apache Flink para ofrecer capacidades de streaming de datos en tiempo real. Para los casos de uso que no son sensibles al tiempo, como la carga de datos históricos, el procesamiento de datos sin conexión o la agregación de registros, se recomienda la importación por lotes. Este método escribe grandes volúmenes de datos en Hologres de una sola vez, lo cual es más eficiente y conserva recursos. Elija entre la importación en tiempo real y la importación por lotes según sus necesidades de negocio y los recursos disponibles. Para obtener más información sobre la importación en tiempo real, consulte Realtime Compute for Apache Flink.

Requisitos previos

Compró una instancia de Hologres. Para obtener más información, consulte Purchase a Hologres instance.
Desplegó un clúster de Apache Flink versión 1.15 o posterior. Para obtener más información, consulte los siguientes temas:
- Apache Flink: Deploy Flink.
- Realtime Compute for Apache Flink: Activate Realtime Compute for Apache Flink.

Importación por lotes con Realtime Compute for Apache Flink

Cree una tabla de resultados de Hologres para almacenar los datos importados desde Flink. Para consultar las instrucciones, vea Connect to HoloWeb and run queries. En este tema se usa la tabla test_sink_customer como ejemplo.
```
-- Create a Hologres result table.
CREATE TABLE test_sink_customer
(
  c_custkey     BIGINT,
  c_name        TEXT,
  c_address     TEXT,
  c_nationkey   INT,
  c_phone       TEXT,
  c_acctbal     NUMERIC(15,2),
  c_mktsegment  TEXT,
  c_comment     TEXT,
  "date"        DATE
) WITH (
  distribution_key="c_custkey,date", 
  orientation="column"
);
```
Nota
Los nombres de campo y los tipos de datos de la tabla de origen de Flink deben coincidir con los de la tabla de resultados de Hologres.

Inicie sesión en la consola de Realtime Compute for Apache Flink. En la página Deployments, haga clic en Create Deployment. Configure los parámetros del despliegue y haga clic en Deploy. Para obtener más información sobre los parámetros, consulte Deploy a JAR job.

A continuación se describen los parámetros clave.

Parámetro	Descripción
Deployment Type	Seleccione JAR.
Deployment Mode	Puede seleccionar el modo de flujo o el modo por lotes. En este tema se utiliza el modo por lotes como ejemplo.
Engine Version	Para obtener más información sobre las versiones del motor, consulte Engine versions y Lifecycle policies. En este tema se usa la versión `vvr-8.0.7-flink-1.17` como ejemplo.
JAR URI	Cargue el conector de Flink de código abierto: hologres-connector-flink-repartition.jar. Nota Puede usar el conector de Flink de código abierto para importar datos por lotes a Hologres. Para consultar el código fuente del conector de Flink, visite el repositorio oficial de Hologres en GitHub.
Entry Point Class	La clase de punto de entrada del programa. La clase principal del conector de Flink es `com.alibaba.ververica.connectors.hologres.example.FlinkToHoloRePartitionExample`.
Entry Point Main Arguments	Proporcione la ruta al archivo `repartition.sql`. En el entorno de ejecución de Realtime Compute for Apache Flink, los archivos de dependencia adicionales se almacenan en `/flink/usrlib/`. Por lo tanto, el argumento completo es `--sqlFilePath="/flink/usrlib/repartition.sql"`.
Additional Dependencies	Cargue el archivo `repartition.sql`. Este archivo es un script de Flink SQL que se usa para definir el origen de datos, declarar la tabla de resultados y configurar la conexión a Hologres. El siguiente código muestra un ejemplo del archivo `repartition.sql`. -- DDL for the source table. This example uses the Flink DataGen connector to generate test data. CREATE TEMPORARY TABLE source_table ( c_custkey BIGINT ,c_name STRING ,c_address STRING ,c_nationkey INTEGER ,c_phone STRING ,c_acctbal NUMERIC(15, 2) ,c_mktsegment STRING ,c_comment STRING ) WITH ( 'connector' = 'datagen' ,'rows-per-second' = '10000' ,'number-of-rows' = '1000000' ); -- DQL for the source table. The query result must match the schema of the result table defined in the sink DDL, including the number and types of fields. SELECT , cast('2024-04-21' as DATE) FROM source_table; -- DDL for the sink table. This declares the result table and configures the connection to Hologres. CREATE TABLE sink_table ( c_custkey BIGINT ,c_name STRING ,c_address STRING ,c_nationkey INTEGER ,c_phone STRING ,c_acctbal NUMERIC(15, 2) ,c_mktsegment STRING ,c_comment STRING ,`date` DATE ) WITH ( 'connector' = 'hologres' ,'dbname' = 'doc_*' ,'tablename' = 'test_sink_customer' ,'username' = 'yourAccessKeyId' ,'password' = 'yourAccessKeySecret' ,'endpoint' = 'hgpostcn-cn-7pp2e1k7-cn-hangzhou.hologres.aliyuncs.com:80' ,'jdbccopywritemode' = 'true' ,'bulkload' = 'true' ,'target-shards.enabled'='true' ); Nota** Para obtener más información sobre los parámetros de conexión de Hologres en el archivo `repartition.sql`, consulte Hologres Flink connector parameters.

Haga clic en el nombre del despliegue y vaya a la página Deployment Details. En la sección Resource Configurations, modifique el valor de Parallelism.

Nota
Se recomienda establecer el paralelismo con el mismo valor que ShardCount de la tabla de resultados de Hologres.
Consulte la tabla de resultados de Hologres.

Una vez enviado el trabajo de Flink, puede consultar los datos escritos en Hologres. Ejemplo de instrucción:
```
SELECT * FROM test_sink_customer;
```

Importación por lotes con Apache Flink

Cree una tabla de resultados de Hologres para almacenar los datos importados desde Flink. Para consultar las instrucciones, vea Connect to HoloWeb and run queries. En este tema se usa la tabla test_sink_customer como ejemplo.
```
-- Create a Hologres result table.
CREATE TABLE test_sink_customer
(
  c_custkey     BIGINT,
  c_name        TEXT,
  c_address     TEXT,
  c_nationkey   INT,
  c_phone       TEXT,
  c_acctbal     NUMERIC(15,2),
  c_mktsegment  TEXT,
  c_comment     TEXT,
  "date"        DATE
) WITH (
  distribution_key="c_custkey,date", 

  orientation="column"
);
```
Nota
Puede establecer el número de fragmentos según su volumen de datos. Para obtener más información sobre los fragmentos, consulte Manage table groups and shard count.

Cree el archivo repartition.sql y cárguelo en cualquier directorio del entorno de su clúster de Flink. En este tema se usa /flink-1.15.4/src/repartition.sql como ruta de ejemplo. El siguiente código muestra un ejemplo del archivo repartition.sql.

Nota

Este archivo es un script de Flink SQL que se usa para definir el origen de datos, declarar la tabla de resultados y configurar la conexión a Hologres.

-- DDL for the source table. This example uses the Flink DataGen connector to generate test data.
CREATE TEMPORARY TABLE source_table
(
  c_custkey     BIGINT
  ,c_name       STRING
  ,c_address    STRING
  ,c_nationkey  INTEGER
  ,c_phone      STRING
  ,c_acctbal    NUMERIC(15, 2)
  ,c_mktsegment STRING
  ,c_comment    STRING
)
WITH (
  'connector' = 'datagen'
  ,'rows-per-second' = '10000'
  ,'number-of-rows' = '1000000'
);

-- DQL for the source table. The query result must match the schema of the result table defined in the sink DDL, including the number and types of fields.
SELECT *, cast('2024-04-21' as DATE) FROM source_table;

-- DDL for the sink table. This declares the result table and configures the connection to Hologres.
CREATE TABLE sink_table
(
  c_custkey     BIGINT
  ,c_name       STRING
  ,c_address    STRING
  ,c_nationkey  INTEGER
  ,c_phone      STRING
  ,c_acctbal    NUMERIC(15, 2)
  ,c_mktsegment STRING
  ,c_comment    STRING
  ,`date`       DATE
)
WITH (
  'connector' = 'hologres'
  ,'dbname' = 'doc_****'
  ,'tablename' = 'test_sink_customer'
  ,'username' = 'yourAccessKeyId'
  ,'password' = 'yourAccessKeySecret'
  ,'endpoint' = 'hgpostcn-cn-7pp2e1k7****-cn-hangzhou.hologres.aliyuncs.com:80'
  ,'jdbccopywritemode' = 'true'
  ,'bulkload' = 'true'
  ,'target-shards.enabled'='true'
);

A continuación se describen los parámetros clave.

Parámetro	Obligatorio	Descripción
connector	Sí	El tipo de conector. El valor debe ser `hologres`.
dbname	Sí	El nombre de la base de datos de Hologres.
tablename	Sí	El nombre de la tabla de Hologres que recibe los datos.
username	Sí	El AccessKey ID de su cuenta de Alibaba Cloud. Puede obtener el AccessKey ID en la página AccessKey Pair.
password	Sí	La contraseña de AccessKey que corresponde a su AccessKey ID.
endpoint	Sí	El endpoint de VPC de la instancia de Hologres. Vaya a la página de detalles de la instancia en la consola de Hologres y obtenga el endpoint en la sección Configurations. Nota El endpoint debe incluir el número de puerto con el formato `ip:port`. Use el endpoint de VPC para conexiones dentro de la misma región. Utilice el endpoint público para conexiones entre regiones.
jdbccopywritemode	No	El método de escritura de datos. Valores válidos: `false` (predeterminado): usa el método `INSERT`. `true`: usa el método `COPY`. Este método incluye `COPY` de streaming (Fixed Copy) y `COPY` por lotes. De forma predeterminada, se utiliza `COPY` de streaming. Nota En comparación con el método `INSERT`, `COPY` de streaming emplea un modelo de streaming para lograr un mayor rendimiento, menor latencia de datos y un consumo reducido de memoria en el cliente, ya que los datos no se agrupan en lotes. Sin embargo, no admite la retracción de datos.
bulkload	No	Indica si se usa el método `COPY` por lotes. Valores válidos: `true`: usa `COPY` por lotes. Esta configuración solo surte efecto cuando `jdbccopywritemode` también se establece en `true`. De lo contrario, se utiliza `COPY` de streaming. Nota Frente a `COPY` de streaming, `COPY` por lotes es más eficiente y aprovecha mejor los recursos de Hologres, lo que se traduce en un rendimiento de escritura superior. Seleccione el método de escritura adecuado según sus requisitos de negocio. Cuando usa `COPY` por lotes para escribir datos en una tabla con clave primaria, pueden producirse bloqueos de tabla. Establezca el parámetro `target-shards.enabled` en `true` para reducir la granularidad del bloqueo del nivel de tabla al nivel de fragmento. Esto permite ejecutar varias tareas de importación por lotes de forma simultánea y reduce la contención por bloqueos de tabla. Comparado con `COPY` de streaming, este enfoque disminuye significativamente la carga en la instancia de Hologres al escribir en una tabla con clave primaria. Las pruebas muestran una reducción de carga de aproximadamente el 66.7 %. Si utiliza `COPY` por lotes y la tabla de destino tiene una clave primaria, la tabla debe estar vacía antes de la operación de escritura. De lo contrario, el proceso se ralentizará debido a la deduplicación de datos basada en la clave primaria. `false` (predeterminado): no usa `COPY` por lotes.
target-shards.enabled	No	Indica si se habilita la escritura por lotes en fragmentos de destino. Valores válidos: `true`: habilita la escritura por lotes en fragmentos de destino. Cuando los datos de origen se reparticionan por fragmento, esta opción reduce la granularidad del bloqueo al nivel de fragmento. `false` (predeterminado): deshabilita esta función.

Nota

Para obtener más información sobre los parámetros de conexión de Hologres en el archivo repartition.sql, consulte Hologres Flink connector parameters.

En el entorno de su clúster de Flink, cargue el conector de Flink de código abierto hologres-connector-flink-repartition.jar en cualquier directorio. En este tema se usa el directorio raíz como ejemplo.

Nota
Puede usar el conector de Flink de código abierto para importar datos por lotes a Hologres. Para consultar el código fuente del conector de Flink, visite el repositorio oficial de Hologres en GitHub.
Envíe el trabajo de Flink. Ejemplo de comando:
```
./bin/flink run -Dexecution.runtime-mode=BATCH -p 3 -c com.alibaba.ververica.connectors.hologres.example.FlinkToHoloRePartitionExample hologres-connector-flink-repartition.jar --sqlFilePath="/flink-1.15.4/src/repartition.sql"
```
Parámetros del comando anterior:
- Dexecution.runtime-mode: el modo de ejecución del trabajo de Flink. Para obtener más información, consulte Execution Mode.
- p: el paralelismo del trabajo. Se recomienda establecer este valor con el mismo número que ShardCount de la tabla de resultados, o bien con un divisor de ShardCount.
- c: el nombre completo de la clase principal en el archivo hologres-connector-flink-repartition.jar.
- sqlFilePath: la ruta al archivo repartition.sql.
Consulte la tabla de resultados de Hologres.

Una vez enviado el trabajo de Flink, puede consultar los datos escritos en Hologres. Ejemplo de instrucción:
```
SELECT * FROM test_sink_customer;
```