Synchronize data from DataHub to Hologres in real time

Limits

• If you enable the IP address whitelist feature for a Hologres instance, you cannot synchronize data from DataHub to the Hologres instance. Do not enable the IP address whitelist feature.
• You must create a child partitioned table in Hologres before you write data to a partitioned table. For more information, see CREATE TABLE.

Synchronization

• Synchronization modes
  • One-by-one insert
    One-by-one insert means that DataHub data is written to Hologres one by one. You must specify the following table property when you create a Hologres table:

    call set_table_property('<table_name>', 'datahub_sync_mode', 'none');

  • Playback
    Playback means the playback of DML operations on the upstream database. DataHub is equivalent to binary logs. If you use the dts-datahub-hologres link to write data, you must set the Apply New Naming Rules of Additional Columns parameter in Data Transmission Service (DTS). You must also specify the following table property when you create a Hologres table:

    • If you set the Apply New Naming Rules of Additional Columns parameter to Yes, you must specify the following table property when you create a Hologres table:

      call set_table_property('<table_name>', 'datahub_sync_mode', 'dts');

    • If you set the Apply New Naming Rules of Additional Columns parameter to No, you must specify the following table property when you create a Hologres table:

      call set_table_property('<table_name>', 'datahub_sync_mode', 'dts_old');

    When DTS synchronizes data to DataHub, eight columns are added to the synchronized data columns to describe the playback data information, such as the INSERT, UPDATE, or DELETE operation. The following part describes the additional columns.

    • Name format of additional columns

      Previous data column name	New data column name
      dts_${Original column name}	${Original column name}

    • Additional column description

      Previous additional column name	New additional column name	Data type	Description
      dts_record_id	new_dts_sync_dts_record_id	String	The unique ID of the incremental log entry.
      dts_operation_flag	new_dts_sync_dts_operation_flag	String	The operation type. Valid values: I: an INSERT operation D: a DELETE operation U: an UPDATE operation
      dts_instance_id	new_dts_sync_dts_instance_id	String	The server ID of the database. The value is set to NULL. The actual value is not displayed to ensure data consistency.
      dts_db_name	new_dts_sync_dts_db_name	String	The name of the database.
      dts_table_name	new_dts_sync_dts_table_name	String	The name of the table.
      dts_utc_timestamp	new_dts_sync_dts_utc_timestamp	String	The operation timestamp, in UTC. It is also the timestamp of the binary log file.
      dts_before_flag	new_dts_sync_dts_before_flag	String	Indicates whether the column values are pre-update values. Valid values: Y and N.
      dts_after_flag	new_dts_sync_dts_after_flag	String	Indicates whether the column values are post-update values. Valid values: Y and N.

• Synchronization policies for primary key conflicts
  If you have set a primary key for a Hologres table, the following two primary key conflict policies are available for the data written from DataHub:

  • Overwrite
    Overwrite means that if a primary key conflict occurs when you write data, new data overwrites original data. In this case, you must specify the following table property when you create a Hologres table:

    call set_table_property('<table_name>', 'datahub_upsert_mode', 'insert_or_replace');

  • Ignore
    Ignore means that if a primary key conflict occurs when you write data, new data is ignored. In other words, the data is not updated, and original data is still used. In this case, you must specify the following table property when you create a Hologres table:

    call set_table_property('<table_name>', 'datahub_upsert_mode', 'insert_or_ignore');

• Combinations of synchronization modes and synchronization policies
  The preceding modes and policies of synchronizing data from DataHub to Hologres can achieve different effects in different combinations. The following part describes the combinations.

  • Combination of the insert mode and the overwrite policy
    This combination is equivalent to executing the following SQL statement in Hologres:

    INSERT INTO target_table (column0,…,columnN) values (?,…,?) ON CONFLICT(PK) DO UPDATE

  • Combination of the insert mode and the ignore policy
    This combination is equivalent to executing the following SQL statement in Hologres:

    INSERT INTO target_table (column0,…,columnN) values (?,…,?) ON CONFLICT(PK) DO NOTHING

  • Combination of the playback mode and the overwrite policy
    • If the value of the dts_operation_flag field is I, this combination is equivalent to executing the following SQL statement in Hologres:

      INSERT INTO target_table (column0,…,columnN) values (?,…,?) ON CONFLICT(PK) DO UPDATE

    • If the value of the dts_operation_flag field is D, this combination is equivalent to executing the following SQL statement in Hologres:

      DELETE FROM target_table where pk=?

    • If the value of the dts_operation_flag field is U and the value of the dts_before_flag field is Y, this combination is equivalent to executing the following SQL statement in Hologres:

      DELETE FROM target_table where pk=?

    • If the value of the dts_operation_flag field is U and the value of the dts_after_flag field is Y, this combination is equivalent to executing the following SQL statement in Hologres:

      INSERT INTO target_table (column0,…,columnN) values (?,…,?) ON CONFLICT(PK) DO UPDATE

  • Combination of the playback mode and the ignore policy
    • If the value of the dts_operation_flag field is I, this combination is equivalent to executing the following SQL statement in Hologres:

      INSERT INTO target_table (column0,…,columnN) values (?,…,?) ON CONFLICT(PK) DO NOTHING

    • If the value of the dts_operation_flag field is D, this combination is equivalent to executing the following SQL statement in Hologres:

      DELETE FROM target_table where pk=?

    • If the value of the dts_operation_flag field is U and the value of the dts_before_flag field is Y, this combination is equivalent to executing the following SQL statement in Hologres:

      DELETE FROM target_table where pk=?

    • If the value of the dts_operation_flag field is U and the value of the dts_after_flag field is Y, this combination is equivalent to executing the following SQL statement in Hologres:

      INSERT INTO target_table (column0,…,columnN) values (?,…,?) ON CONFLICT(PK) DO NOTHING

Procedure

1. Prepare a data source in DataHub.
   1. Create a DataHub project.
      1. Log on to the DataHub console. In the left-side navigation pane, click Projects.
      2. On the Projects page, click Create Project in the upper-right corner.
      3. In the Create Project panel, set the parameters as required and click Create.

      
   2. Create a DataHub topic.
      1. On the Projects page, find the created project and click the project name.
      2. In the upper-right corner of the project details page, click Create Topic. In the Create Topic panel, set the parameters as required.

      

      Parameter	Description
      Creation Type	Create Directly: Create a topic. Import MaxCompute Tables: Select an existing table in MaxCompute to create a topic.
      Name	The name of the custom topic.
      Type	The type of the topic. Valid values: TUPLE: structured data BLOB: unstructured data
      Schema Details	The details of the schema. The Schema Details parameter is displayed if you set the Type parameter to TUPLE. You can create fields based on your business requirements. If you select Allow Null for a field and the field does not exist in the upstream, the field is set to NULL. If you clear Allow Null for a field, the field configuration is strictly verified. An error is returned if the type of written data does not match the data type specified for the field.
      Number of Shards	The number of shards in the topic. Shards ensure the concurrent data transmission of a topic. Each shard has a unique ID. A shard may be in one of the following states: Opening: The shard is starting. Active: The shard is started and available. Each active shard consumes server resources. We recommend that you create shards as needed.
      Lifecycle	The period that each record can be retained in the topic. Unit: days. Valid values: 1 to 7.
      Description	The description of the topic.

   3. Write data to the topic.
      Use a tool, such as Realtime Compute for Apache Flink, or an application to write data to the created topic.
2. Create a Hologres table to which the data is to be synchronized.
   Create a Hologres table used to receive data. You must create fields with data types corresponding to those of the DataHub topic.

   The following table describes the mappings between data types in DataHub and Hologres.

   DataHub	Hologres
   BIGINT	BIGINT
   STRING	TEXT
   BOOLEAN	BOOLEAN
   DOUBLE	DOUBLE PRECISION
   TIMESTAMP	TIMESTAMPTZ
   DECIMAL	DECIMAL

   For example, you can execute the following statements to create a table:

   BEGIN;
   CREATE TABLE lineitem ( 
   L_ORDERKEY BIGINT NOT NULL,
   L_PARTKEY BIGINT NOT NULL,
   L_SUPPKEY BIGINT NOT NULL,
   L_LINENUMBER BIGINT NOT NULL,
   L_QUANTITY DECIMAL(20,10),
   L_EXTENDEDPRICE DECIMAL(20,10),
   L_DISCOUNT DECIMAL(20,10),
   L_TAX DECIMAL(20,10),
   L_RETURNFLAG TEXT,
   L_LINESTATUS TEXT,
   L_SHIPDATE TIMESTAMPTZ,
   L_COMMITDATE TIMESTAMPTZ,
   L_RECEIPTDATE TIMESTAMPTZ,
   L_SHIPINSTRUCT TEXT,
   L_SHIPMODE TEXT,
   L_COMMENT TEXT
   );
   
   CALL set_table_property('lineitem', 'orientation', 'column');
   CALL set_table_property('lineitem', 'datahub_sync_mode', 'none');
   CALL set_table_property('lineitem', 'datahub_upsert_mode', 'insert_or_ignore');
   
   COMMIT;

3. Create a DataConnector in DataHub to synchronize data to Hologres.
   1. On the Topic List tab of the project details page, find the created topic and click the topic name.
   2. On the details page of the topic, click Connector in the upper-right corner.
   3. In the Create Connector panel, click Hologres. Then, set the parameters as required and click Create.

      Parameter	Description
      Instance	The ID of the Hologres instance. You can log on to the Hologres console and copy the ID of the required Hologres instance.
      Database	The name of the Hologres database to which you want to write data.
      Table	The name of the Hologres table to which you want to write data.
      Import Fields	The fields to be synchronized to the Hologres table. You can synchronize all or part of the fields of the DataHub topic based on your business requirements.
      Authentication Mode	The mode in which the access to the Hologres instance is authenticated. Default value: AccessKey.
      AccessKey ID	The AccessKey ID of the current Alibaba Cloud account used to access the Hologres instance. You can obtain the AccessKey ID from the Security Management page.
      AccessKey Secret	The AccessKey secret of the current Alibaba Cloud account used to access the Hologres instance. You can obtain the AccessKey secret from the Security Management page.
      Timestamp Unit	The timestamp unit used by the DataConnector for data synchronization. Valid values: MICROSECOND: microseconds. This is the default value. MILLISECOND: milliseconds. SECOND: seconds.

4. Wait until the data is synchronized to Hologres.
   On the topic details page, click the Connectors tab. Find the created DataConnector and view the status of the DataConnector. The status indicates the progress of the data synchronization.
5. Query the synchronized data in Hologres.
   Connect the Hologres instance to a development tool and use the tool to check whether the data is synchronized to the Hologres instance in real time. For more information about development tools, see Overview. For example, you can execute the following statement to query the synchronized data:

   SELECT COUNT(*) FROM lineitem;

Common errors and troubleshooting

This section describes common errors that are reported when you use Hologres to help you troubleshoot the errors.

• Scenario 1: The following error message is returned when you query data:

  ErrorMessage [Import field not found in dest schema;

  Possible cause: The datahub_sync_mode property of the Hologres table is not set to dts.

  Solution: Create another Hologres table and set the table property datahub_sync_mode to dts.

• Scenario 2: The following error message is returned when you query data:

  ErrorCode=InternalServerError; ErrorMessage =Field already exists 

  Possible cause: The datahub_sync_mode property of the Hologres table is set to dts, and eight additional columns are included when you create the table.

  Solution: Create another Hologres table. If you set the datahub_sync_mode property to dts, keep the fields the same as those of the upstream. You do not need to add eight additional columns.