Lindorm:Utiliser Flink pour accéder à LindormTable

Déterminer le type de table à accéder

Dans Lindorm, utilisez Lindorm SQL pour identifier le type de LindormTable qu'un job Flink en cours de développement doit accéder :

1. Utilisez lindorm-cli (ou directement Lindorm Insight ou DMS) pour vous connecter au moteur de tables larges de Lindorm.

2. Exécutez l'instruction SQL suivante pour vérifier l'attribut IS_HBASE_LIKE de la table.

   • Si la valeur de l'attribut est TRUE, il s'agit d'une table HBase.

   • Si la valeur de l'attribut est FALSE, il s'agit d'une table SQL.

   SHOW TABLE VARIABLES FROM table_name LIKE 'IS_HBASE_LIKE';

Choisir le type de connecteur

Selon la forme du produit Flink que vous avez sélectionnée, décidez quel connecteur utiliser pour accéder à LindormTable.

Obtenir les informations de connexion de LindormTable

• Scénario 1 : Utilisation du connecteur HBase open source ou du connecteur Lindorm pour base de données multi-modèles cloud native Dans ce scénario, l'adresse de connexion doit être l'adresse d'accès à l'API Java HBase (VPC) du moteur de tables larges. Sur la page des détails de l'instance Lindorm, cliquez sur Database connections dans le menu de gauche, puis sélectionnez l'onglet Wide Table Engine. Dans la zone Connect through HBase-compatible address, récupérez l'adresse VPC au format ld-<instance-ID>-proxy-lindorm.lindorm.rds.aliyuncs.com:30020. Dans la zone Connect through MySQL-compatible address, récupérez l'adresse compatible MySQL au format ld-<instance-ID>-proxy-sql-lindorm.lindorm.rds.aliyuncs.com:33060.

• Scénario 2 : Utilisation du connecteur JDBC Dans ce scénario, l'adresse de connexion doit être l'adresse compatible MySQL (VPC) du moteur de tables larges. Sur la page des détails de l'instance Lindorm, cliquez sur Database connections dans le volet de navigation de gauche, puis sélectionnez l'onglet Wide Table Engine. Dans la zone Connect through HBase-compatible address, vous pouvez consulter l'adresse VPC de l'API Java HBase (au format ld-<instance-ID>-proxy-lindorm.rds.aliyuncs.com:30020).

Prérequis

• Développer un job utilisant Flink communautaire pour accéder à LindormTable

  • Pour utiliser le connecteur HBase open source afin d'accéder à LindormTable, assurez-vous que la version du moteur de tables larges est ultérieure ou égale à 2.4.3.

  • Pour utiliser le connecteur JDBC open source afin d'accéder à LindormTable, assurez-vous que la version du moteur de tables larges est ultérieure ou égale à 2.6.5.2 et que vous avez activé la fonctionnalité de compatibilité MySQL.

  • Pour savoir comment afficher ou mettre à niveau la version actuelle, consultez les rubriques Notes de version de LindormTable et Mettre à niveau la version mineure du moteur d'une instance Lindorm.

• Si vous utilisez Realtime Compute for Apache Flink pour développer un job accédant à LindormTable, aucune restriction de version ne s'applique au moteur de tables larges.

• Assurez-vous que l'environnement où se trouve le cluster Flink dispose d'une connectivité réseau avec l'instance Lindorm et que les adresses IP du client ont été ajoutées à la liste blanche de Lindorm. Pour savoir comment ajouter des adresses IP à la liste blanche, consultez la rubrique Configurer une liste blanche.

Développement de jobs avec Flink communautaire

# Create Flink table and start the job
CREATE TABLE source_table(
    c1 INT,
    c2 STRING
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '2',
  'fields.c2.length' = '5',
  'fields.c1.min' = '1',
  'fields.c1.max' = '100'
);

CREATE TABLE sink_table(
    c1 INT,
    c2 STRING
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://ld-xxxxx-proxy-lindorm.lindorm.rds.aliyuncs.com:33060/default?sslMode=disabled&allowPublicKeyRetrieval=true&useServerPrepStmts=true&useLocalSessionState=true&rewriteBatchedStatements=true&cachePrepStmts=true&prepStmtCacheSize=300&prepStmtCacheSqlLimit=50000000',
    'username' = 'root',
    'password' = 'root',
    'table-name' = 'testflink'
);

INSERT INTO sink_table SELECT * FROM source_table;

Développement de jobs avec Realtime Compute for Apache Flink