Accéder aux tables DLF Iceberg via l'API Java - Data Lake Formation

Cette rubrique explique comment accéder au catalogue REST DLF Iceberg à l'aide de l'API Java Apache Iceberg open source, ainsi qu'avec Spark et Flink. Vous pouvez gérer les métadonnées des tables et lire ou écrire dans les tables Iceberg via le protocole REST Iceberg standard avec signature AWS SigV4. Le plan de données utilise S3FileIO, conforme aux standards de la communauté, pour se connecter directement aux endpoints compatibles S3 d'OSS.

Prérequis

Votre environnement d'exécution doit utiliser JDK 17 ou une version ultérieure (requis par Iceberg 1.11.0).
Ajoutez les dépendances communautaires Apache Iceberg suivantes (version 1.11.0) à votre projet Maven. Toutes les dépendances sont disponibles sur Maven Central :

<dependencies>
    <dependency>
        <groupId>org.apache.iceberg</groupId>
        <artifactId>iceberg-core</artifactId>
        <version>1.11.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.iceberg</groupId>
        <artifactId>iceberg-aws</artifactId>
        <version>1.11.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.iceberg</groupId>
        <artifactId>iceberg-aws-bundle</artifactId>
        <version>1.11.0</version>
    </dependency>
</dependencies>

Remarque

Si vous devez lire et écrire des données Parquet directement via l'API Java (sans moteur de calcul), ajoutez également les dépendances iceberg-data, iceberg-parquet, iceberg-orc, parquet-hadoop et hadoop-common.

Se connecter via l'API Java

Utilisez la classe RESTCatalog, conforme aux standards de la communauté, pour vous connecter au catalogue REST DLF Iceberg. Transmettez les paramètres de configuration suivants lors de l'initialisation :

Map<String, String> props = new HashMap<>();
props.put("uri", "http://cn-hangzhou-vpc.dlf.aliyuncs.com/iceberg");
props.put("warehouse", "iceberg_table_test");
props.put("rest.auth.type", "sigv4");
props.put("rest.auth.sigv4.delegate-auth-type", "none");
props.put("rest.signing-region", "cn-hangzhou");
props.put("rest.signing-name", "DlfNext");
props.put("rest.access-key-id", "xxx");
props.put("rest.secret-access-key", "yyy");
props.put("io-impl", "org.apache.iceberg.aws.s3.S3FileIO");
RESTCatalog icebergCatalog = new RESTCatalog();
icebergCatalog.initialize(catalogName, props);

Paramètres de configuration

Paramètre

Description

Exemple

uri

Endpoint du catalogue REST DLF Iceberg (accessible au sein du VPC). Le format est http://<region>-vpc.dlf.aliyuncs.com/iceberg. Pour obtenir la liste complète des endpoints de service par région, consultez la section Régions et endpoints.

http://cn-hangzhou-vpc.dlf.aliyuncs.com/iceberg

warehouse

Nom du catalogue DLF.

iceberg_test

rest.auth.type

Type d'authentification. Définissez la valeur sur sigv4.

sigv4

rest.auth.sigv4.delegate-auth-type

Type d'authentification déléguée. Définissez la valeur sur none.

none

rest.signing-region

ID de la région de l'instance DLF.

cn-hangzhou

rest.signing-name

Nom du service de signature. Définissez la valeur sur DlfNext.

DlfNext

rest.access-key-id

ID AccessKey pour l'accès à DLF.

rest.secret-access-key

Secret AccessKey pour l'accès à DLF.

io-impl

Classe d'implémentation File IO. Utilise S3FileIO, conforme aux standards de la communauté. Définissez la valeur sur org.apache.iceberg.aws.s3.S3FileIO.

org.apache.iceberg.aws.s3.S3FileIO

Accès avec Apache Spark

Apache Spark open source peut lire et écrire directement dans les tables DLF Iceberg en utilisant le runtime communautaire Iceberg. Le catalogue communique via le protocole REST Iceberg standard avec signature AWS SigV4, tandis que le plan de données utilise S3FileIO (conforme aux standards de la communauté) pour se connecter aux endpoints compatibles S3 d'OSS via l'API compatible S3. Toutes les dépendances proviennent de Maven Central.

Démarrer Spark SQL

spark-sql \
  --packages org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.11.0,org.apache.iceberg:iceberg-aws-bundle:1.11.0 \
  --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
  --conf spark.sql.catalog.dlf=org.apache.iceberg.spark.SparkCatalog \
  --conf spark.sql.catalog.dlf.type=rest \
  --conf spark.sql.catalog.dlf.uri=http://cn-hangzhou-vpc.dlf.aliyuncs.com/iceberg \
  --conf spark.sql.catalog.dlf.warehouse=iceberg_table_test \
  --conf spark.sql.catalog.dlf.rest.auth.type=sigv4 \
  --conf spark.sql.catalog.dlf.rest.auth.sigv4.delegate-auth-type=none \
  --conf spark.sql.catalog.dlf.rest.signing-region=cn-hangzhou \
  --conf spark.sql.catalog.dlf.rest.signing-name=DlfNext \
  --conf spark.sql.catalog.dlf.rest.access-key-id=xxx \
  --conf spark.sql.catalog.dlf.rest.secret-access-key=yyy \
  --conf spark.sql.catalog.dlf.io-impl=org.apache.iceberg.aws.s3.S3FileIO

Lire et écrire avec Spark SQL

Une fois la configuration terminée, utilisez Spark SQL standard pour lire et écrire dans les tables Iceberg :

CREATE DATABASE IF NOT EXISTS dlf.demo_db;
CREATE TABLE dlf.demo_db.demo_tbl (id BIGINT, name STRING) USING iceberg;
INSERT INTO dlf.demo_db.demo_tbl VALUES (1, 'hello'), (2, 'world');
SELECT * FROM dlf.demo_db.demo_tbl;

Accès avec Apache Flink

Apache Flink open source utilise le runtime communautaire Iceberg pour accéder aux tables DLF Iceberg. Placez les fichiers iceberg-flink-runtime-1.20-1.11.0.jar et iceberg-aws-bundle-1.11.0.jar (artifacts communautaires disponibles sur Maven Central) dans le répertoire lib/ de votre installation Flink, puis enregistrez le catalogue dans le client SQL.

Enregistrer le catalogue

CREATE CATALOG dlf WITH (
  'type' = 'iceberg',
  'catalog-type' = 'rest',
  'uri' = 'http://cn-hangzhou-vpc.dlf.aliyuncs.com/iceberg',
  'warehouse' = 'iceberg_table_test',
  'rest.auth.type' = 'sigv4',
  'rest.auth.sigv4.delegate-auth-type' = 'none',
  'rest.signing-region' = 'cn-hangzhou',
  'rest.signing-name' = 'DlfNext',
  'rest.access-key-id' = 'xxx',
  'rest.secret-access-key' = 'yyy',
  'io-impl' = 'org.apache.iceberg.aws.s3.S3FileIO'
);

Lire et écrire avec Flink SQL

USE CATALOG dlf;
CREATE DATABASE IF NOT EXISTS demo_db;
CREATE TABLE demo_db.demo_tbl (id BIGINT, name STRING);
INSERT INTO demo_db.demo_tbl VALUES (1, 'hello'), (2, 'world');
SELECT * FROM demo_db.demo_tbl;