Realtime Compute for Apache Flink:Connecteur Apache Iceberg

Apache Iceberg est un format de table de lac de données ouvert. Vous pouvez utiliser Apache Iceberg pour créer rapidement votre propre service de stockage de lac de données sur HDFS ou sur OSS cloud, et exploiter les moteurs de calcul de l'écosystème big data open source, tels que Flink, Spark, Hive et Presto, pour effectuer des analyses de lac de données.

Le connecteur Apache Iceberg offre les fonctionnalités suivantes :

• Crée un service de stockage de lac de données léger et peu coûteux basé sur HDFS ou sur le stockage d'objets.

• Fournit une sémantique ACID complète.

• Prend en charge les requêtes de voyage dans le temps pour accéder aux versions historiques des données.

• Prend en charge le filtrage efficace des données.

• Prend en charge l'évolution du schéma.

• Prend en charge l'évolution du partitionnement.

CREATE TABLE iceberg_table (
  id    BIGINT,
  data  STRING
  PRIMARY KEY(`id`) NOT ENFORCED
)
 PARTITIONED BY (data)
 WITH (
 'connector' = 'iceberg',
  ...
);

Assurez-vous de disposer d'un compartiment OSS et d'une base de données Data Lake Formation (DLF). Pour plus d'informations, consultez les sections Créer un compartiment et Bases de données, tables et fonctions.

Cet exemple utilise le connecteur Datagen pour générer des données en continu aléatoires et les écrire dans une table Iceberg.

CREATE TEMPORARY TABLE datagen(
  id    BIGINT,
  data  STRING
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE dlf_iceberg (
  id    BIGINT,
  data  STRING
) WITH (
  'connector' = 'iceberg',
  'catalog-name' = '<yourCatalogName>',
  'catalog-database' = '<yourDatabaseName>',
  'io-impl' = 'org.apache.iceberg.aliyun.oss.OSSFileIO',
  'oss.endpoint' = '<yourOSSEndpoint>',  
  'access.key.id' = '${secret_values.ak_id}',
  'access.key.secret' = '${secret_values.ak_secret}',
  'catalog-impl' = 'org.apache.iceberg.aliyun.dlf.DlfCatalog',
  'warehouse' = '<yourOSSWarehousePath>',
  'dlf.catalog-id' = '<yourCatalogId>',
  'dlf.endpoint' = '<yourDLFEndpoint>',  
  'dlf.region-id' = '<yourDLFRegionId>'
);

INSERT INTO dlf_iceberg SELECT * FROM datagen;

Ingestion de données

Vous pouvez utiliser le connecteur Apache Iceberg comme destination (sink) dans un job YAML pour l'ingestion de données.

Syntaxe

sink:
  type: iceberg
  name: Iceberg Sink
  catalog.properties.rest.signing-region: cn-beijing
  catalog.properties.uri: http://cn-beijing-vpc.dlf.aliyuncs.com/iceberg
  catalog.properties.warehouse: flink_iceberg
  catalog.properties.type: rest
  catalog.properties.io-impl: org.apache.iceberg.rest.DlfFileIO

Paramètres

Réutiliser un catalogue existant

À partir de VVR 11.5, vous pouvez référencer directement un catalogue Iceberg intégré créé sur la page Gestion des données dans un job d'ingestion de données Flink CDC. Cela simplifie votre configuration en réduisant le nombre de propriétés de connexion requises.

sink:
  type: iceberg
  using.built-in-catalog: iceberg_catalog

Les jobs d'ingestion de données peuvent réutiliser automatiquement tous les paramètres du catalogue Iceberg. Cela équivaut à configurer manuellement les paramètres avec le préfixe catalog.properties. dans votre job YAML.

Si vous souhaitez remplacer les paramètres réutilisés, vous pouvez spécifier explicitement les paramètres YAML correspondants. Ces derniers sont prioritaires.

Exemple

L'exemple suivant montre comment utiliser un catalogue DLF comme catalogue Iceberg et écrire des données dans Data Lake Formation (DLF) :

• Pour plus d'informations sur les paramètres préfixés par catalog.properties, consultez la section Créer un catalogue DLF Iceberg.

  source:
    type: mysql
    name: MySQL Source
    hostname: ${secret_values.mysql.hostname}
    port: ${mysql.port}
    username: ${secret_values.mysql.username}
    password: ${secret_values.mysql.password}
    tables: ${mysql.source.table}
    server-id: 8601-8604
  
  sink:
    type: iceberg
    name: Iceberg Sink
    catalog.properties.rest.signing-region: cn-beijing
    catalog.properties.uri: http://cn-beijing-vpc.dlf.aliyuncs.com/iceberg
    catalog.properties.warehouse: flink_iceberg
    catalog.properties.type: rest
    catalog.properties.io-impl: org.apache.iceberg.rest.DlfFileIO

Modifications de schéma

Lorsqu'il est utilisé comme destination (sink) d'ingestion de données, le connecteur Apache Iceberg prend en charge les modifications de schéma suivantes :

• CREATE TABLE

• ADD COLUMN

• ALTER COLUMN TYPE (la modification du type d'une colonne de clé primaire n'est pas prise en charge)

• RENAME COLUMN

• DROP COLUMN

• TRUNCATE TABLE

• DROP TABLE