Hologres:Replicar dados do Simple Log Service

Etapa 1: Criar uma tabela no Hologres

Crie uma tabela de destino no Hologres. Adicione índices aos campos que você pretende consultar frequentemente para melhorar o desempenho. Para obter todas as opções de Data Definition Language (DDL), consulte Criar uma tabela.

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

Etapa 2: Gravar dados usando o Flink

Crie um job Flink SQL com uma tabela source do SLS e uma tabela sink do Hologres e insira dados entre elas.

Para obter informações básicas sobre os conectores usados aqui:

• Conector da tabela source do SLS: Tabela source do SLS

• Conector da tabela sink do Hologres: Tabela sink do Hologres

Como o Flink não tem um tipo JSON nativo, o campo content usa VARCHAR como substituto. Os dados JSON são gravados diretamente no Hologres, que os armazena como JSONB. Alternativamente, analise o JSON no Flink antes da gravação.

Definir a tabela source do SLS:

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

Definir a tabela sink do Hologres:

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

Inserir dados:

Todos os campos de metadados da source do SLS (__topic__, __source__, __timestamp__, __tag__) são mapeados para as colunas correspondentes do Hologres. O timestamp Unix __timestamp__ é convertido para TIMESTAMP, e __receive_time__ é extraído do mapa de tags e convertido para 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;

Para obter etapas sobre como desenvolver e executar jobs SQL no Flink, consulte Mapa de desenvolvimento de jobs e Iniciar um job.

Etapa 3: Verificar dados no Hologres

Após o início do job Flink, consulte a tabela de destino para confirmar o fluxo de dados:

Etapa 1: Criar uma tabela no Hologres

Crie uma tabela de destino no Hologres. Neste exemplo, uid atua como chave primária (garante a unicidade das linhas) e como chave de distribuição (direciona linhas com o mesmo uid para o mesmo shard, melhorando o desempenho da consulta). O campo timestamp é definido como a coluna de tempo do evento.

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;

Para mais opções de DDL, consulte Criar uma tabela.

Etapa 2: Configurar fontes de dados

Adicione ambas as fontes de dados ao seu workspace do DataWorks antes de criar uma tarefa de sincronização.

• SLS: adicione uma fonte de dados LogHub. Consulte Configurar uma fonte de dados LogHub (SLS).

• Hologres: adicione uma fonte de dados do Hologres. Consulte Configurar uma fonte de dados do Hologres.

Etapa 3: Criar uma tarefa de sincronização em tempo real

Crie e execute uma tarefa de sincronização em tempo real na integração de dados do DataWorks. Defina a entrada como a fonte de dados LogHub e a saída como a fonte de dados do Hologres. Configure o mapeamento de campos conforme mostrado abaixo:

Para obter etapas detalhadas, consulte Configurar uma tarefa de sincronização em tempo real para dados incrementais em uma única tabela.

Etapa 4: Verificar dados no Hologres

Após o início da tarefa de sincronização, consulte a tabela de destino para confirmar o fluxo de dados: