AnalyticDB:Importer des données depuis Apache Flink

Télécharger les fichiers JAR requis

Connecteur Flink JDBC

Téléchargez le fichier JAR du connecteur correspondant à votre version de Flink. Pour les versions non répertoriées ci-dessous, consultez la documentation JDBC SQL Connector.

Pilote MySQL

Téléchargez un pilote MySQL (version 5.1.40 ou ultérieure) depuis le dépôt mysql/mysql-connector-java.

Placez les deux fichiers JAR dans le répertoire lib de chaque nœud Flink, puis redémarrez le cluster. Pour obtenir des instructions sur le démarrage du cluster, consultez la rubrique Step 2: Start a cluster.

Compatibilité des versions pour les anciennes versions de Flink

Pour les versions de Flink antérieures à la 1,11 :

• Flink 1.9 et 1.10 : Connecteur JDBC Flink v1.10

• Flink 1.8 et versions antérieures : Flink v1.8 JDBCAppendTableSink

Créer la table source

Créez une table source Flink qui lit les données depuis le fichier CSV :

CREATE TABLE IF NOT EXISTS csv_person (
  `user_id`   STRING,
  `user_name` STRING,
  `age`       INT
) WITH (
  'connector'              = 'filesystem',
  'path'                   = 'file:///root/data.csv',
  'format'                 = 'csv',
  'csv.ignore-parse-errors' = 'true',
  'csv.allow-comments'     = 'true'
);

Créer la table de résultat

Créez une table de résultat Flink qui correspond à la table de destination dans AnalyticDB for MySQL :

CREATE TABLE mysql_person (
  user_id   STRING,
  user_name STRING,
  age       INT
) WITH (
  'connector'                  = 'jdbc',
  'url'                        = 'jdbc:mysql://<endpoint>:<port>/<db_name>?useServerPrepStmts=false&rewriteBatchedStatements=true',
  'table-name'                 = '<table_name>',
  'username'                   = '<username>',
  'password'                   = '<password>',
  'sink.buffer-flush.max-rows' = '10',
  'sink.buffer-flush.interval' = '1s'
);

Remplacez les espaces réservés suivants avant d'exécuter cette instruction :

Paramètres du connecteur

Pour connaître toutes les options disponibles du connecteur, consultez la rubrique Connector options.

Comportement de vidange

Lorsque les paramètres sink.buffer-flush.max-rows et sink.buffer-flush.interval sont tous deux définis sur des valeurs non nulles, une vidange est déclenchée dès que l'une des conditions suivantes est remplie :

• Le nombre de lignes mises en tampon atteint la valeur de sink.buffer-flush.max-rows.

• Le temps écoulé depuis la dernière vidange atteint la valeur de sink.buffer-flush.interval.

Exécuter l'importation

Exécutez l'instruction suivante pour commencer à écrire les données de la table source vers AnalyticDB for MySQL :

INSERT INTO mysql_person SELECT user_id, user_name, age FROM csv_person;

SELECT * FROM person;