Realtime Compute for Apache Flink:Create a Hologres result table

Background information

• Introduction to Hologres

  Hologres is a real-time interactive analytics service that is developed by Alibaba Cloud. Hologres is compatible with the PostgreSQL protocol and is seamlessly integrated with the big data ecosystem. Hologres allows you to analyze and process up to petabytes of data with high concurrency and low latency. Hologres allows you to use existing business intelligence (BI) tools to perform multidimensional analysis and explore your business in an efficient manner.

• Features
  The following table describes the features that are supported by the Hologres connector.

  Feature	Description
  Streaming semantics	Allows the system to write changelogs to a Hologres result table.
  Merge data into a wide table	Updates only data in specific fields instead of data in an entire row.
  Create Hologres tables as result tables in the CREATE TABLE AS and CREATE DATABASE AS statements	Synchronizes data from a database or a single table in real time and synchronizes schema changes of each table to the related Hologres table in real time during database synchronization or table synchronization.

Prerequisites

A Hologres table is created. For more information, see Manage an internal table.

Limits

• Only Realtime Compute for Apache Flink that uses Ververica Runtime (VVR) 2.0.0 or later supports the Hologres connector.
• You cannot use the Hologres connector to write result data to Hologres external tables. For more information about Hologres external tables, see Manage a foreign table.

DDL syntax

CREATE TABLE hologres_sink(
  name varchar,
  age BIGINT,
  birthday BIGINT,
  PRIMARY KEY (name) NOT ENFORCED
) WITH (
  'connector'='hologres',
  'dbname'='<yourDbname>',
  'tablename'='<yourTablename>',
  'username'='<yourUsername>',
  'password'='<yourPassword>',
  'endpoint'='<yourEndpoint>',
  'field_delimiter'='|' -- This parameter is optional. 
);

Parameters in the WITH clause

Streaming semantics

Stream processing, which is also known as streaming data or streaming event processing, refers to the continuous processing of a series of unbounded data or events. In most cases, the system that processes streaming data or streaming events allows you to specify a reliability pattern or processing semantics to ensure data accuracy. Network or device failures may cause data loss.

Merge data into a wide table

If you want to write data from multiple streaming deployments to one Hologres wide table, you can merge the data into a wide table. For example,

one Flink data stream contains fields A, B, and C, and the other contains fields A, D, and E. The Hologres wide table WIDE_TABLE contains fields A, B, C, D, and E, among which field A is the primary key. You can perform the following operations:

1. Execute Flink SQL statements to create two Hologres result tables. One table is used to declare fields A, B, and C, and the other table is used to declare fields A, D, and E. Map the two result tables to the Hologres wide table named WIDE_TABLE.
2. Parameter settings of the two Hologres result tables:
   • Set mutatetype to insertorupdate. This indicates that data is updated based on the primary key.
   • Set ignoredelete to true. This prevents retraction messages from generating Delete requests.
3. Insert data from the two Flink data streams into the two result tables.

Create Hologres tables as result tables in the CREATE TABLE AS and CREATE DATABASE AS statements

You can execute the CREATE TABLE AS or CREATE DATABASE AS statements to synchronize data from a single table or an entire database in real time. The changes to the schema of the source table can also be synchronized to the related Hologres table in real time during database or table synchronization.

If the schema of the source table is changed, Flink synchronizes the changes to the table schema to the related Hologres table and then writes data to the Hologres table. Flink automatically completes this process. For more information, see CREATE TABLE AS statement, CREATE DATABASE AS statement, and Ingest data into data warehouses in real time.

Data type mappings

Sample code

CREATE TEMPORARY TABLE datagen_source(
  name varchar,
  age BIGINT,
  birthday BIGINT
) WITH (
  'connector'='datagen'
);

CREATE TEMPORARY TABLE hologres_sink (
  name varchar,
  age BIGINT,
  birthday BIGINT
) WITH (
  'connector'='hologres',
  'dbname'='<yourDbname>',
  'tablename'='<yourTablename>',
  'username'='<yourAccessID>',
  'password'='<yourAccessSecret>',
  'endpoint'='<yourEndpoint>'
);

INSERT INTO hologres_sink SELECT * from datagen_source;