Use a connector to write data from Flink to Hologres

CREATE TABLE mysource(
  name varchar,
  age BIGINT,
  birthday BIGINT
) WITH (
  'connector.type'='hologres',
  'connector.database'='Name of the Hologres database to be connected',
  'connector.table'='Name of the Hologres table used to receive data',
  'connector.username'='AccessKey ID of the account used to connect to Hologres',
  'connector.password'='AccessKey secret of the account used to connect to Hologres',
  'connector.endpoint'='<ip>:<port>'// The endpoint and port number used to connect to Hologres.
);

• Streaming sinking semantics

  If you want to write data from a Flink sink to Hologres in a continuous manner and query the data written to Hologres in real time, use the streaming sinking semantics. We recommend that you use the streaming sinking semantics to write streaming data from a Flink sink to Hologres in real time.

  You can use the exactly-once or at-least-once semantic based on the configurations of the sink and attributes of the Hologres table.

  • Exactly-once: The system processes data or events only once, no matter whether a fault occurs.
  • At-least-once: If data or an event is lost before a process is completed, the system transfers all data and events anew. In this case, data or events may be processed multiple times. If the first retry succeeds, no other retries are required.
  When you use streaming sinking semantics in Flink sinks, pay attention to the following points:

  • If the destination Hologres table has a primary key, you can ensure the idempotence based on the primary key conflicts to implement the exactly-once semantics.
    When data entries with the same primary key are written to Hologres, you can set mutationType to one of the following values to ensure that each data entry is written exactly once to Hologres:

    • insertorignore: Retain each data entry written to Hologres for the first time and ignore all data entries with the same primary key value written to Hologres later. This is the default value.
    • insertorreplace: Update the existing data entry as a whole.
    • insertorupdate: Update the specified fields of the existing data entry.
  • If the destination Hologres table contains no primary key, implement the at-least-once semantics.

  By default, you can only import data from a Flink sink to a non-partitioned table or a child partitioned table in Hologres. If you import data to a parent partitioned table, no error message is returned but you cannot query the written data in Hologres.

  To route data in a parent partitioned table to the corresponding child partitioned tables, set the partitionRouter parameter to true. For example, you can use the following statement to create a Flink sink, write data to a parent partitioned table, and then route data from the parent partitioned table to the corresponding child partitioned tables:

  CREATE TABLE mysource(
    name varchar,
    age BIGINT,
    birthday BIGINT
  ) WITH (
    'connector.type'='hologres',
    'connector.database'='Name of the Hologres database to be connected',
    'connector.table'='Name of the parent partitioned table in Hologres used to receive data',
    'connector.username'='AccessKey ID of the account used to connect to Hologres',
    'connector.password'='AccessKey secret of the account used to connect to Hologres',
    'connector.endpoint'='<ip>:<port>',// The endpoint and port number used to connect to Hologres.
    'connector.partitionRouter'='true',
  );

  Note

  • Set the tableName parameter to the name of the parent partitioned table.
  • Hologres does not automatically create child partitioned tables when it receives data from Flink. You must create child partitioned tables as needed in advance.
• Batch sinking semantics

  If you want to write a large amount of data entries from a Flink sink to Hologres at a time, use the batch sinking semantics. You can query the data only after all the data is written to Hologres. The data writing process runs as a transaction. If the transaction succeeds, all the data is written to Hologres exactly once.

  The batch sinking semantics improves the high-throughput performance. We recommend that you use the batch sinking semantics when you process multiple data entries in Flink at a time, such as data migration and data backfilling.

  Note To use the batch sinking semantics to write data from a Flink sink to a partitioned table in Hologres, you can only write the data to a child partitioned table.

CREATE TABLE mysource(
  name varchar,
  age BIGINT,
  birthday BIGINT
) WITH (
  'connector.type'='hologres',
  'connector.database'='Name of the Hologres database to be connected',
  'connector.table'='Name of the Hologres table used to receive data',
  'connector.username'='AccessKey ID of the account used to connect to Hologres',
  'connector.password'='AccessKey secret of the account used to connect to Hologres',
  'connector.endpoint'='<ip>:<port>'// The endpoint and port number used to connect to Hologres.
);

CREATE TABLE mysource(
  ...
) WITH (
  'connector.type'='hologres',
  'connector.database'='Name of the Hologres database to be connected',
  'connector.table'='Name of the Hologres table used to receive data',
  'connector.username'='AccessKey ID of the account used to connect to Hologres',
  'connector.password'='AccessKey secret of the account used to connect to Hologres',
  'connector.endpoint'='<ip>:<port>'// The endpoint and port number used to connect to Hologres.
);

// register hologreslookup as udf
SELECT * FROM source, LATERAL TABLE(hologreslookup(a, b))

// register hologreslookup as lookup table source
SELECT x, a, b, c FROM src 
    JOIN hologreslookup FOR SYSTEM_TIME AS OF src.proc as h ON src.x = h.a AND src.y = h.b