Hologres:Hologres sink table

Limitations

• Different versions of Blink in exclusive mode use different development syntaxes. Before you start, identify your version of Blink in exclusive mode and use the corresponding examples.

• Ensure that your Realtime Compute for Apache Flink deployment and Hologres instance are in the same region to avoid connection failures.

• Versions of Blink in exclusive mode earlier than 3.6 do not have a built-in Hologres connector. To write data to Hologres in real time, you must reference a JAR file. For assistance, see Common errors during upgrade preparation or contact us by joining the Hologres DingTalk group. For more information, see How can I get more online support?.

  Note

  We recommend upgrading Blink in exclusive mode to version 3.6 or later to run your jobs.

• Blink exclusive mode 3.7 supports the automatic creation of Hologres partitioned tables, but you must configure createparttable='true' in the job. The considerations for using partitioned tables are as follows:

  • Hologres currently supports only list partitioning.

  • When you create a partitioned table, you must explicitly specify the partition key column. Currently, partition key columns support only the text and int4 data types, and their values cannot contain a hyphen (-), for example, 2020-09-12.

  • If a primary key is defined for the partitioned table, the partition key column must be part of the primary key.

  • When you create a child partition table, the value of its partition key column must be a fixed value.

  • The value of the partition key column for data written to a child partition table must exactly match the value defined when the child table was created. Otherwise, an error occurs.

  • The DEFAULT partition feature is not currently supported.

• If the destination Hologres table has a primary key, the default real-time write semantic does not update records based on that key. If you later import data with a duplicate primary key, the new data is discarded.

• Hologres writes data asynchronously. You must add the blink.checkpoint.fail_on_checkpoint_error=true configuration to your job to ensure that a failover is triggered if the job encounters an exception. This parameter is not required for Blink 3.7.6 or later.

DDL syntax

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.

WITH parameters

Write to a standard Hologres sink table

1. Create a table in Hologres.

   Create a table in Hologres to receive the data. The following is an example SQL statement:

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

2. Create a Realtime Compute for Apache Flink job.

   1. Log on to the Realtime Compute for Apache Flink console.

   2. Create a job.

      • Blink in exclusive mode version 3.6 and later includes built-in support for the Hologres data source. You can use this data source directly. The following is an example SQL statement:

        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. Publish the job.

   1. After you create the job, click Syntax Check. A Successful status indicates that the syntax is correct.

   2. Click Save to save the job.

   3. Click Publish to deploy the job to the production environment. Configure the deployment settings based on your business requirements. Click Publish New Version to start the deployment process. In the Initial Resources step, choose a resource allocation method: Auto-tuning based on last run, System allocation, or Manual resource configuration. After you make your selection, click Next. You can also click Skip Data Check to go directly to the resource configuration step.

4. Start the job.

   After publishing the job to the production environment, you must start it manually.

   In the top navigation bar on the Development Platform page, click Administration on the right. On the Administration page, select the desired job and click Start in the upper-right corner.

5. Query data in Hologres in real time.

   Query the destination table in Hologres to view the written data in real time. The following is a sample query:

   select * from blink_test;

Merge and update wide tables

This section describes a common use case: writing data from multiple streams into a single Hologres wide table.

Write to a partitioned Hologres sink table

Hologres allows you to call the HoloHub API to write data directly to a parent partitioned table. The data is then automatically routed to the correct child partition tables. For more information, see HoloHub API.

1. Create a partitioned table in Hologres.

   Create a partitioned table in Hologres to receive data and create its corresponding child partition tables. The following is a sample SQL statement:

   -- 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;

   Note

   • When you run the command, replace the ${bizdate} parameter with the actual value.

   • Only version 3.7 or later of Blink in exclusive mode supports automatic partition creation. If you are using an earlier version, you must create child partition tables in Hologres in advance. Otherwise, data import will fail.

2. Create a job in Blink in exclusive mode.

   The following is a sample statement for creating a job in Blink in exclusive mode.

   Note

   The following example applies to Blink exclusive mode 3.7 and later. If you are using a version of Blink exclusive mode earlier than 3.7, upgrade to version 3.7 or later, or remove the `createparttable` = 'true' configuration.

   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. Publish and start the job.

   For more information, refer to the Publish the job and Start the job steps in the Write data to a standard Hologres sink table section.

4. Query data in Hologres in real time.

   Query the destination table in Hologres to view the written data in real time. The following are example queries:

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

Data type mappings

See Data type summary for data type mappings between Blink in exclusive mode and Hologres.