Hologres:Replicate data from Simple Log Service

Step 1: Create a Hologres table

Create a destination table in Hologres. Add indexes for fields you plan to query frequently to improve performance. For full Data Definition Language (DDL) options, see Create a table.

CREATE TABLE sls_flink_holo (
    content JSONB,
    operation TEXT,
    uid TEXT,
    topic  TEXT,
    source TEXT,
    c__timestamp TIMESTAMPTZ,
    receive_time BIGINT,
    PRIMARY KEY (uid)
);

Step 2: Write data using Flink

Create a Flink SQL job with an SLS source table and a Hologres sink table, then insert data between them.

For background on the connectors used here:

• SLS source table connector: SLS source table

• Hologres sink table connector: Hologres sink table

Because Flink does not have a native JSON type, the content field uses VARCHAR as a substitute. The JSON data is written directly to Hologres, which stores it as JSONB. Alternatively, parse the JSON in Flink before writing.

Define the SLS source table:

CREATE TEMPORARY TABLE sls_input (
    content STRING,
    operation STRING,
    uid STRING,
    `__topic__` STRING METADATA VIRTUAL,
    `__source__` STRING METADATA VIRTUAL,
    `__timestamp__` BIGINT METADATA VIRTUAL,
    `__tag__` MAP<VARCHAR, VARCHAR> METADATA VIRTUAL
)
WITH (
    'connector' = 'sls',
    'endpoint' = '<sls-private-endpoint>',    -- Private endpoint of the SLS project
    'accessid' = '<your-access-key-id>',      -- AccessKey ID
    'accesskey' = '<your-access-key-secret>', -- AccessKey secret
    'starttime' = '2024-08-30 00:00:00',      -- Start time to consume logs
    'project' = '<your-sls-project>',         -- SLS project name
    'logstore' = '<your-logstore>'            -- Logstore name
);

Define the Hologres sink table:

CREATE TEMPORARY TABLE hologres_sink (
    content VARCHAR,
    operation VARCHAR,
    uid VARCHAR,
    topic  STRING,
    source STRING,
    c__timestamp TIMESTAMP,
    receive_time BIGINT
)
WITH (
    'connector' = 'hologres',
    'dbname' = '<your-holo-database>',       -- Hologres database name
    'tablename' = '<your-holo-table>',       -- Hologres destination table name
    'username' = '<your-access-key-id>',     -- AccessKey ID
    'password' = '<your-access-key-secret>', -- AccessKey secret
    'endpoint' = '<your-holo-vpc-endpoint>'  -- VPC endpoint of the Hologres instance
);

Insert data:

All metadata fields from the SLS source (__topic__, __source__, __timestamp__, __tag__) are mapped to the corresponding Hologres columns. The __timestamp__ Unix timestamp is cast to TIMESTAMP, and __receive_time__ is extracted from the tag map and cast to BIGINT.

INSERT INTO hologres_sink
SELECT
   content,
   operation,
   uid,
   `__topic__`,
   `__source__`,
   CAST(FROM_UNIXTIME(`__timestamp__`) AS TIMESTAMP),
   CAST(__tag__['__receive_time__'] AS BIGINT) AS receive_time
FROM
   sls_input;

For steps on developing and running SQL jobs in Flink, see Job development map and Start a job.

Step 3: Verify data in Hologres

After the Flink job starts, query the destination table to confirm data is flowing in:

Step 1: Create a Hologres table

Create a destination table in Hologres. In this example, uid is the primary key (ensures row uniqueness) and the distribution key (routes rows with the same uid to the same shard, improving query performance). The timestamp field is set as the event time column.

BEGIN;
CREATE TABLE sls_dw_holo (
    content JSONB,
    operation TEXT,
    uid TEXT,
    C_Topic  TEXT,
    C_Source TEXT,
    timestamp BIGINT,
    PRIMARY KEY (uid)
);
CALL set_table_property('sls_dw_holo', 'distribution_key', 'uid');
CALL set_table_property('sls_dw_holo', 'event_time_column', 'timestamp');
COMMIT;

For more DDL options, see Create a table.

Step 2: Configure data sources

Add both data sources to your DataWorks workspace before creating a sync task.

• SLS: Add a LogHub data source. See Configure a LogHub (SLS) data source.

• Hologres: Add a Hologres data source. See Configure a Hologres data source.

Step 3: Create a real-time sync task

Create and run a real-time sync task in DataWorks data integration. Set the input to the LogHub data source and the output to the Hologres data source. Configure field mapping as shown below:

For detailed steps, see Configure a real-time synchronization task for incremental data in a single table.

Step 4: Verify data in Hologres

After the sync task starts, query the destination table to confirm data is flowing in: