Hologres:Hologres sink table

Limitaciones

• Cada versión de Blink en modo exclusivo utiliza una sintaxis de desarrollo distinta. Antes de comenzar, identifique la versión de Blink en modo exclusivo que usa y utilice los ejemplos correspondientes.

• Asegúrese de que el despliegue de Realtime Compute for Apache Flink y la instancia de Hologres estén en la misma región para evitar errores de conexión.

• Las versiones de Blink en modo exclusivo anteriores a la 3.6 no incluyen un connector de Hologres integrado. Para escribir datos en Hologres en tiempo real, debe hacer referencia a un archivo JAR. Si necesita ayuda, consulte Common errors during upgrade preparation o contáctenos al unirte al grupo de DingTalk de Hologres. Para obtener más información, consulte How can I get more online support?.

  Nota

  Se recomienda actualizar Blink en modo exclusivo a la versión 3.6 o posterior para ejecutar sus jobs.

• Blink en modo exclusivo 3.7 admite la creación automática de tablas particionadas de Hologres, pero debe configurar createparttable='true' en el job. Tenga en cuenta las siguientes consideraciones al usar tablas particionadas:

  • Actualmente, Hologres solo admite el particionamiento por lista.

  • Al crear una tabla particionada, debe especificar de forma explícita la columna de clave de partición. Por ahora, estas columnas solo admiten los tipos de datos text e int4, y sus valores no pueden contener un guion (-); por ejemplo, 2020-09-12.

  • Si la tabla particionada tiene definida una clave primaria, la columna de clave de partición debe formar parte de dicha clave.

  • Cuando cree una tabla de partición secundaria, el valor de su columna de clave de partición debe ser un valor fijo.

  • El valor de la columna de clave de partición de los datos escritos en una tabla de partición secundaria debe coincidir exactamente con el valor definido al crear dicha tabla secundaria. De lo contrario, se producirá un error.

  • Por ahora, no se admite la función de partición DEFAULT.

• Si la tabla de destino de Hologres tiene una clave primaria, la semántica predeterminada de escritura en tiempo real no actualiza los registros según esa clave. Si después importa datos con una clave primaria duplicada, los datos nuevos se descartan.

• Hologres escribe los datos de forma asíncrona. Agregue la configuración blink.checkpoint.fail_on_checkpoint_error=true a su job para garantizar que se active un failover si el job presenta una excepción. Este parámetro no es obligatorio en Blink 3.7.6 o posterior.

Sintaxis DDL

create table Hologres_sink(
  name varchar,
  age BIGINT,
  birthday BIGINT
) with (
  type='hologres',
  dbname='<yourDbname>', -- The name of the Hologres database.
  tablename='<yourTablename>', -- The name of the Hologres table that receives the data.
  username='<yourUsername>', -- The AccessKey ID of your Alibaba Cloud account.
  password='<yourPassword>', -- The AccessKey secret of your Alibaba Cloud account.
  endpoint='<yourEndpoint>'); -- The VPC endpoint of your Hologres instance.

Parámetros WITH

Escritura en una tabla sink estándar de Hologres

1. Cree una tabla en Hologres.

   Cree una tabla en Hologres para recibir los datos. A continuación se muestra un ejemplo de instrucción SQL:

    create table blink_test (a int, b text, c text, d float8, e bigint);

2. Cree un job de Realtime Compute for Apache Flink.

   1. Inicie sesión en la consola de Realtime Compute for Apache Flink.

   2. Cree un job.

      • Blink en modo exclusivo 3.6 y versiones posteriores incluye compatibilidad integrada con el origen de datos de Hologres. Puede utilizar este origen de datos directamente. A continuación se muestra un ejemplo de instrucción SQL:

        create table randomSource (a int, b VARCHAR, c VARCHAR, d DOUBLE, e BIGINT) with (type = 'random');
        create table test (
          a int,
          b VARCHAR,
          c VARCHAR,
          PRIMARY KEY (a)
        ) with (
          type = 'hologres',
          `endpoint` = '$ip:$port', -- The VPC endpoint and port number of your Hologres instance.
          `username` = 'The AccessKey ID of your Alibaba Cloud account',
          `password` = 'The AccessKey secret of your Alibaba Cloud account',
          `dbname` = 'The name of the Hologres database',
          `tablename` = 'blink_test'-- The name of the Hologres table that receives the data.
        );
        insert
          into test
        select
          a,b,c
        from
          randomSource;

3. Publique el job.

   1. Después de crear el job, haga clic en Syntax Check. Un estado Successful indica que la sintaxis es correcta.

   2. Haga clic en Save para guardar el job.

   3. Haga clic en Publish para desplegar el job en el entorno de producción. Configure los ajustes de despliegue según sus necesidades empresariales. Haga clic en Publish New Version para iniciar el proceso de despliegue. En el paso Initial Resources, elija un método de asignación de recursos: Auto-tuning based on last run, System allocation o Manual resource configuration. Una vez hecha la selección, haga clic en Next. También puede hacer clic en Skip Data Check para ir directamente al paso de configuración de recursos.

4. Inicie el job.

   Después de publicar el job en el entorno de producción, debe iniciarlo manualmente.

   En la barra de navegación superior de la página Development Platform, haga clic en Administration a la derecha. En la página Administration, seleccione el job deseado y haga clic en Start uicontrol替换 修改时间:2026-02-03 01:28:40 原文:启动 修改后:conref="mcms_dita_hologres-console.dita#hologres-console/start" en la esquina superior derecha.

5. Consulte los datos en Hologres en tiempo real.

   Consulte la tabla de destino en Hologres para ver los datos escritos en tiempo real. A continuación se muestra un ejemplo de consulta:

   select * from blink_test;

Fusión y actualización de tablas anchas

En esta sección se describe un caso de uso común: escribir datos de varios flujos en una única tabla ancha de Hologres.

Escritura en una tabla sink particionada de Hologres

Hologres permite llamar a la API de HoloHub para escribir datos directamente en una tabla particionada principal. Después, los datos se enrutan automáticamente a las tablas de partición secundarias correctas. Para obtener más información, consulte HoloHub API.

1. Cree una tabla particionada en Hologres.

   Cree una tabla particionada en Hologres para recibir los datos y genere sus tablas de partición secundarias correspondientes. A continuación se muestra un ejemplo de instrucción SQL:

   -- Create the parent table test_message and its child partition tables.
   drop table if exists test_message;
   begin;
   create table test_message (
    "bizdate" text NOT NULL,
    "tag" text NOT NULL,
    "id" int4 NOT NULL,
    "title" text NOT NULL,
    "body" text,
   PRIMARY KEY (bizdate,tag,id)
   )
   PARTITION BY LIST (bizdate);
   commit;

   Nota

   • Cuando ejecute el comando, reemplace el parámetro ${bizdate} por el valor real.

   • Solo Blink en modo exclusivo 3.7 o posterior admite la creación automática de particiones. Si utiliza una versión anterior, debe crear las tablas de partición secundarias en Hologres por adelantado. De lo contrario, la importación de datos fallará.

2. Cree un job en Blink en modo exclusivo.

   A continuación se muestra un ejemplo de instrucción para crear un job en Blink en modo exclusivo.

   Nota

   El siguiente ejemplo aplica a Blink en modo exclusivo 3.7 y versiones posteriores. Si utiliza una versión de Blink en modo exclusivo anterior a la 3.7, actualice a la versión 3.7 o posterior, o bien elimine la configuración createparttable.

   create table test_message_src(
     tag VARCHAR,
     id INTEGER,
     title VARCHAR,
     body VARCHAR
   ) with (
     type = 'random',
     `interval` = '10',
     `count` = '100'
   );
   create table test_message_sink (
     bizdate VARCHAR,
     tag VARCHAR,
     id INTEGER,
     title VARCHAR,
     body VARCHAR
   ) with (
     type = 'hologres',
     `endpoint` = '$ip:$port', -- The VPC endpoint of your Hologres instance.
     `username` ='<AccessID>', -- The AccessKey ID of your Alibaba Cloud account.
     `password` = '<AccessKey>', -- The AccessKey secret of your Alibaba Cloud account.
     `dbname` = '<DBname>', -- The name of the Hologres database.
     `tablename` = '<Tablename>', -- The name of the table in your Hologres database.
     `partitionrouter` = 'true', -- Write data to a partitioned table in Hologres.
     `createparttable` = 'true' -- Automatically create child partition tables in Hologres.
   );
   insert into test_message_sink select "20200327",* from test_message_src;
   insert into test_message_sink select "20200328",* from test_message_src;

3. Publique e inicie el job.

   Para obtener más información, consulte los pasos Publish the job y Start the job de la sección Write data to a standard Hologres sink table.

4. Consulte los datos en Hologres en tiempo real.

   Consulte la tabla de destino en Hologres para ver los datos escritos en tiempo real. A continuación se muestran algunos ejemplos de consultas:

   select * from test_message;
   select * from test_message where bizdate = '20200327';

Asignaciones de tipos de datos

Consulte Data type summary para ver las asignaciones de tipos de datos entre Blink en modo exclusivo y Hologres.