Hologres:Consumir dados binlog do Hologres com o Flink

Antes de consumir dados binlog do Hologres, observe o seguinte:

• O consumo de dados binlog requer o Hologres V0.9 ou posterior. A configuração de uma lista de permissões de engine requer o Hologres V1.3.21 ou posterior. Ao ativar uma lista de permissões em uma instância com versão anterior ao Hologres V1.3.21, o consumo de binlog falhará. Caso sua versão seja anterior à exigida, entre em contato com Obtain online support for Hologres para obter assistência.

• O Hologres oferece suporte ao recurso binlog no nível de tabela para tabelas orientadas a linhas e tabelas orientadas a colunas. A partir do Hologres V1.1, tabelas de armazenamento híbrido linha-coluna também passaram a ser suportadas. Após ativar o recurso binlog, tabelas orientadas a colunas têm, teoricamente, overhead maior do que tabelas orientadas a linhas. Portanto, em cenários com atualizações frequentes de dados, recomendamos ativar o recurso binlog para tabelas orientadas a linhas.

• Para mais informações sobre o suporte a binlog e sobre como ativar e configure esse recurso, consulte Subscribe to Hologres binlogs.

• Somente o Realtime Compute for Apache Flink fornecido pela Alibaba Cloud oferece suporte ao consumo de dados binlog do Hologres. No modo HoloHub, o consumo de dados binlog do Hologres com o Flink oferece suporte apenas a tipos de dados simples. A partir do Flink VVR 6.0.3, é possível consumir dados binlog do Hologres no modo JDBC. Em comparação com o modo HoloHub, o modo JDBC oferece suporte a mais tipos de dados. Para mais informações, consulte Data type mappings between Blink/Flink and Hologres. Esse modo exige permissões adicionais. Para detalhes, consulte Permissions.

• Não é possível consumir dados binlog de uma tabela particionada pai.

• O Hologres V2.0 oferece suporte limitado ao modo HoloHub. A partir do V2.1, o modo HoloHub foi descontinuado e totalmente substituído pelo modo JDBC. Antes de atualizar sua instância do Hologres, consulte Switch from HoloHub mode to JDBC mode para verificar os jobs do Flink que utilizam o modo HoloHub, atualize-os para o Flink VVR adequado e, em seguida, atualize sua instância do Hologres.

Neste modo, os dados binlog consumidos pelo Source são repassados aos nós downstream como dados regulares do Flink. Isso significa que todos os dados são tratados como dados do tipo Insert, e você pode escolha como processar dados de um tipo específico de hg_binlog_event_type conforme os requisitos do seu negócio. Após ativar o binlog para uma tabela do Hologres, utilize o seguinte DDL para uma tabela source (no modo não-CDC) no Flink para consumir o binlog em tempo real.

create table test_message_src_binlog_table(
  hg_binlog_lsn BIGINT,
  hg_binlog_event_type BIGINT,
  hg_binlog_timestamp_us BIGINT,
  id INTEGER,
  title VARCHAR,
  body VARCHAR
) with (
  'connector'='hologres',
  'dbname'='<yourDbname>',
  'tablename'='<yourTablename>',
  'username'='<yourAccessID>',
  'password'='<yourAccessSecret>',
  'endpoint'='<yourEndpoint>',
  'binlog' = 'true',
  'binlogMaxRetryTimes' = '10',
  'binlogRetryIntervalMs' = '500',
  'binlogBatchReadSize' = '100'
);

• Os três parâmetros binlogxxx são campos de sistema do binlog. Seus nomes e tipos são fixos e não podem ser modificados.

• Os demais campos correspondem aos campos definidos pelo usuário e devem estar todos em letras minúsculas.

Neste modo, o Source consome os dados binlog e define automaticamente o tipo correto de Flink RowKind (INSERT, DELETE, UPDATE_BEFORE ou UPDATE_AFTER) para cada linha com base no hg_binlog_event_type. Isso possibilita a sincronização espelhada dos dados da tabela, semelhante ao recurso CDC do MySQL e do PostgreSQL.

Após ativar o recurso binlog para uma tabela do Hologres, utilize o seguinte DDL no Flink para crie uma tabela source e consumir dados binlog em tempo real no modo CDC.

create table test_message_src_binlog_table(
  id INTEGER,
  title VARCHAR,
  body VARCHAR
) with (
  'connector'='hologres',
  'dbname'='<yourDbname>',
  'tablename'='<yourTablename>',
  'username'='<yourAccessID>',
  'password'='<yourAccessSecret>',
  'endpoint'='<yourEndpoint>',
  'binlog' = 'true',
  'cdcMode' = 'true',
  'binlogMaxRetryTimes' = '10',
  'binlogRetryIntervalMs' = '500',
  'binlogBatchReadSize' = '100'
);

Migrar do modo HoloHub para o modo JDBC

O Hologres começou a descontinuar o modo HoloHub na V2.0. Caso precise atualizar sua instância do Hologres, migre seus jobs do modo HoloHub para o modo JDBC conforme descrito abaixo.

Se você tiver muitos jobs do Flink VVR que consomem dados binlog do Hologres, utilize o método a seguir para identificar os jobs e as tabelas que precisam ser atualizados.

1. Baixe a ferramenta open-source find-incompatible-flink-jobs-1,0-SNAPSHOT-jar-with-dependencies.jar.

2. Na linha de comando local, acesse o diretório da ferramenta open-source e execute o seguinte comando para visualize todos os jobs e tabelas que precisam ser atualizados.

   Nota

   Para execute o comando a seguir, é necessário ter um ambiente Java instalado, com JDK 8 ou posterior.

   java -cp find-incompatible-flink-jobs-1.0-SNAPSHOT-jar-with-dependencies.jar com.alibaba.hologres.FindIncompatibleFlinkJobs <region> <url> <AccessKeyID> <AccessKeySecret> <binlog/rpc>
   # Example
   java -cp find-incompatible-flink-jobs-1.0-SNAPSHOT-jar-with-dependencies.jar com.alibaba.hologres.FindIncompatibleFlinkJobs Beijing https://vvp.console.alibabacloud.com/web/xxxxxx/en/#/workspaces/xxxx/namespaces/xxxx/operations/stream/xxxx my-access-key-id my-access-key-secret binlog

   A tabela a seguir descreve os parâmetros.

   Parâmetro	Descrição
   region	A região onde seu workspace do Realtime Compute for Apache Flink está localizado. Para a lista de valores válidos, consulte a tabela Region values.
   url	A URL de qualquer job no workspace de destino do Realtime Compute for Apache Flink.
   AccessKey ID	O AccessKey ID de uma conta com permissão para acessar o workspace do Realtime Compute for Apache Flink.
   AccessKey secret	O AccessKey secret de uma conta com permissão para acessar o workspace do Realtime Compute for Apache Flink.
   binlog/rpc	O tipo de conteúdo do job a ser verificado. Valores válidos: binlog: Verifica todas as tabelas source de binlog do Hologres em todos os jobs do workspace. rpc: Verifica todas as tabelas de dimensão ou tabelas de resultado que utilizam o modo rpc em todos os jobs do workspace.

   Valores de região (clique para expandir)

   Região	Valor
   China (Beijing)	Beijing
   China (Shanghai)	Shanghai
   China (Hangzhou)	Hangzhou
   China (Shenzhen)	Shenzhen
   China (Zhangjiakou)	Zhangjiakou
   China (Hong Kong)	Hong Kong
   Singapore	Singapore
   Germany (Frankfurt)	Frankfurt
   Indonesia (Jakarta)	Jakarta
   Malaysia (Kuala Lumpur)	Kuala Lumpur
   US (Silicon Valley)	us-west-1
   China East 2 Finance	China East 2 Finance

3. O código a seguir apresenta um exemplo de resultado.

   dwd_xxx nx not a running job, skip
   dwd_st xxx not a running job, skip
   dwd_xxx a not a running job, skip
   dia xxx not a hologres job, skip
   dwd_tr xxx ail not sql job, skip
   --- The following are Hologres binlog source tables that are on a version earlier than 8.0.5 and do not have sdkmode = jdbc set ---
   deploymentName                         version                              tableName
   ads_xxx                   xxx vvr-6.0.7-flink-1.15 adsdb.ads_xxx              xxx         xxx catalog
          xxx      vvr-8.0.1-flink-1.17 odsdb.ods_xxx      nts
   xxx           vvr-8.0.1-flink-1.17 dwsdb.dws_xxx             tag
   xxx       s vvr-8.0.1-flink-1.17 odsdb.ods_xxx        ers