Hologres:Consumir dados de binlog do Hologres com o Flink

Antes de consumir dados de binlog do Hologres, observe os seguintes pontos:

• O consumo de dados de binlog está disponível apenas no Hologres V0.9 ou posterior. A configuração de uma lista de permissões de engine é suportada somente a partir do Hologres V1.3.21. Se você 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 a versão da sua instância seja inferior à necessária, Obtenha suporte online para o Hologres para receber assistência.

• O Hologres oferece suporte ao recurso de binlog no nível de tabela para tabelas orientadas a linhas e tabelas orientadas a colunas. A partir do Hologres V1.1, tabelas com armazenamento híbrido de linhas e colunas também são suportadas. Após ativar o recurso de binlog, as tabelas orientadas a colunas teoricamente apresentam maior sobrecarga do que as orientadas a linhas. Por isso, em cenários com atualizações frequentes de dados, recomendamos ativar o recurso de binlog em tabelas orientadas a linhas.

• Para obter mais informações sobre o suporte a binlog e como ativar e configurar esse recurso, consulte Assinar binlogs do Hologres.

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

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

• O Hologres V2.0 oferece suporte limitado ao modo HoloHub. A partir da versão V2.1, o modo HoloHub foi descontinuado e totalmente substituído pelo modo JDBC. Antes de atualizar sua instância do Hologres, consulte Migrar do modo HoloHub para o modo JDBC para verificar seus jobs do Flink que utilizam o modo HoloHub, atualize os jobs do Flink VVR conforme necessário e, em seguida, atualize a instância do Hologres.

Neste modo, os dados de binlog consumidos pela source são repassados aos nós downstream como dados comuns do Flink. Isso significa que todos os dados são tratados como tipo Insert, permitindo que você escolha como processar dados de um tipo específico de hg_binlog_event_type conforme suas necessidades de negócio. Após ativar o binlog em uma tabela do Hologres, utilize o seguinte DDL para criar uma tabela de origem (no modo não CDC) no Flink e 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 alterados.

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

Neste modo, a source consome dados de binlog e define automaticamente o tipo RowKind correto do Flink (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 PostgreSQL.

Após ativar o recurso de binlog em uma tabela do Hologres, utilize o seguinte DDL no Flink para criar uma tabela de origem e consumir dados de 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 iniciou a descontinuação do modo HoloHub na versão V2.0. Se você precisar atualizar sua instância do Hologres, deverá migrar seus jobs do modo HoloHub para o modo JDBC conforme descrito abaixo.

Caso possua diversos jobs do Flink VVR que consomem dados de binlog do Hologres, utilize o método abaixo para identificar os jobs e tabelas que precisam ser atualizados.

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

2. Em uma linha de comando local, acesse o diretório da ferramenta open-source e execute o comando a seguir para visualizar todos os jobs e tabelas que requerem atualização.

   Nota

   Para executar o comando abaixo, é necessário ter um ambiente Java instalado, com JDK 8 ou superior.

   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.aliyun.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 uma lista de valores válidos, consulte a tabela Valores de região.
   url	A URL de qualquer job no workspace de destino do Realtime Compute for Apache Flink.
   AccessKeyID	O AccessKeyID de uma conta com permissão para acessar o workspace do Realtime Compute for Apache Flink.
   AccessKeySecret	O AccessKeySecret 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 de origem de binlog do Hologres em todos os jobs dentro do workspace. rpc: Verifica todas as tabelas de dimensão ou tabelas de resultado que usam o modo rpc em todos os jobs dentro do workspace.

   Valores de região (clique em 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
   Singapura	Singapore
   Alemanha (Frankfurt)	Frankfurt
   Indonésia (Jakarta)	Jakarta
   Malásia (Kuala Lumpur)	Kuala Lumpur
   EUA (Silicon Valley)	us-west-1
   China East 2 Finance	China East 2 Finance

3. O código a seguir mostra 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