Hologres:Consommer les données binlog de Hologres avec Flink

Avant de consommer les données binlog de Hologres, tenez compte des points suivants :

• La consommation des données binlog n'est possible qu'à partir de Hologres V0.9 ou version ultérieure. La configuration d'une liste blanche du moteur est disponible uniquement pour Hologres V1.3.21 ou version ultérieure. Si vous activez une liste blanche sur une instance exécutant une version antérieure à Hologres V1.3.21, la consommation du binlog échoue. Si la version de votre instance est antérieure à celle requise, obtenez une assistance en ligne pour Hologres.

• Hologres prend en charge la fonctionnalité binlog au niveau de la table pour les tables orientées ligne et les tables orientées colonne. À partir de Hologres V1.1, les tables à stockage hybride ligne-colonne sont également prises en charge. Après l'activation de la fonctionnalité binlog, les tables orientées colonne entraînent théoriquement une surcharge plus élevée que les tables orientées ligne. Par conséquent, pour les scénarios impliquant des mises à jour fréquentes des données, nous vous recommandons d'activer la fonctionnalité binlog pour les tables orientées ligne.

• Pour plus d'informations sur la prise en charge du binlog et sur la manière d'activer et de configurer cette fonctionnalité, consultez la rubrique S'abonner aux binlogs Hologres.

• Seule la solution Realtime Compute for Apache Flink fournie par Alibaba Cloud permet de consommer les données binlog de Hologres. En mode HoloHub, la consommation des données binlog de Hologres avec Flink ne prend en charge que les types de données simples. À partir de Flink VVR 6.0.3, vous pouvez consommer les données binlog de Hologres en mode JDBC. Par rapport au mode HoloHub, le mode JDBC prend en charge un plus grand nombre de types de données. Pour plus d'informations, consultez la rubrique Mappages des types de données entre Blink/Flink et Hologres. Ce mode nécessite des autorisations supplémentaires. Pour plus de détails, consultez la section Autorisations.

• Vous ne pouvez pas consommer les données binlog d'une table partitionnée parente.

• Hologres V2.0 offre une prise en charge limitée du mode HoloHub. À partir de la version V2.1, le mode HoloHub est obsolète et entièrement remplacé par le mode JDBC. Avant de mettre à niveau votre instance Hologres, reportez-vous à la section Passer du mode HoloHub au mode JDBC afin de vérifier vos jobs Flink utilisant actuellement le mode HoloHub, de mettre à niveau vos jobs Flink VVR en conséquence, puis de mettre à niveau votre instance Hologres.

Dans ce mode, les données binlog consommées par la source sont transmises aux nœuds en aval sous forme de données Flink standard. Cela signifie que toutes les données sont traitées comme des données de type Insert, et vous pouvez choisir comment traiter les données d'un type hg_binlog_event_type spécifique en fonction de vos besoins métier. Une fois le binlog activé pour une table Hologres, vous pouvez utiliser la DDL suivante pour une table source (en mode non-CDC) dans Flink afin de consommer le binlog en temps réel.

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

• Les trois paramètres binlogxxx sont des champs système binlog. Leurs noms et types sont fixes et ne peuvent pas être modifiés.

• Les autres champs correspondent à vos champs définis par l'utilisateur et doivent être écrits entièrement en minuscules.

Dans ce mode, la source consomme les données binlog et définit automatiquement le type Flink RowKind correct (INSERT, DELETE, UPDATE_BEFORE ou UPDATE_AFTER) pour chaque ligne en fonction du champ hg_binlog_event_type. Cela permet une synchronisation miroir des données de la table, similaire à la fonctionnalité CDC de MySQL et PostgreSQL.

Après avoir activé la fonctionnalité binlog pour une table Hologres, vous pouvez utiliser la DDL suivante dans Flink pour créer une table source permettant de consommer les données binlog en temps réel en mode 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'
);

Passer du mode HoloHub au mode JDBC

Hologres a commencé à abandonner le mode HoloHub à partir de la version V2.0. Si vous devez mettre à niveau votre instance Hologres, vous devez faire passer vos jobs du mode HoloHub au mode JDBC comme décrit ci-dessous.

Si vous disposez de nombreux jobs Flink VVR consommant les données binlog de Hologres, vous pouvez utiliser la méthode suivante pour identifier les jobs et les tables nécessitant une mise à niveau.

1. Téléchargez l'outil open source find-incompatible-flink-jobs-1,0-SNAPSHOT-jar-with-dependencies.jar.

2. Dans une ligne de commande locale, accédez au répertoire de l'outil open source et exécutez la commande suivante pour afficher tous les jobs et tables nécessitant une mise à niveau.

   Remarque

   Pour exécuter la commande suivante, vous devez disposer d'un environnement Java installé, avec JDK 8 ou version ultérieure.

   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

   Le tableau suivant décrit les paramètres.

   Paramètre	Description
   region	La région où se trouve votre espace de travail Realtime Compute for Apache Flink. Pour la liste des valeurs valides, consultez le tableau Valeurs de région.
   url	L'URL de n'importe quel job dans l'espace de travail Realtime Compute for Apache Flink cible.
   AccessKey ID	L'AccessKey ID d'un compte disposant des autorisations nécessaires pour accéder à l'espace de travail Realtime Compute for Apache Flink.
   AccessKey secret	L'AccessKey secret d'un compte disposant des autorisations nécessaires pour accéder à l'espace de travail Realtime Compute for Apache Flink.
   binlog/rpc	Le type de contenu de job à vérifier. Valeurs valides : binlog : Vérifie toutes les tables sources binlog Hologres dans tous les jobs de l'espace de travail. rpc : Vérifie toutes les tables de dimension ou tables de résultats utilisant le mode rpc dans tous les jobs de l'espace de travail.

   Valeurs de région (cliquez pour développer)

   Région	Valeur
   Chine (Pékin)	Beijing
   Chine (Shanghai)	Shanghai
   Chine (Hangzhou)	Hangzhou
   Chine (Shenzhen)	Shenzhen
   Chine (Zhangjiakou)	Zhangjiakou
   Chine (Hong Kong)	Hong Kong
   Singapour	Singapore
   Allemagne (Francfort)	Frankfurt
   Indonésie (Jakarta)	Jakarta
   Malaisie (Kuala Lumpur)	Kuala Lumpur
   États-Unis (Silicon Valley)	us-west-1
   Chine Est 2 Finance	China East 2 Finance

3. Le code suivant fournit un exemple de résultat.

   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