Java API 访问 DLF Iceberg 表-数据湖构建(DLF)-阿里云帮助中心

本文介绍如何通过开源 Apache Iceberg Java API、Spark 和 Flink 访问 DLF Iceberg REST Catalog，管理表元数据并读写 Iceberg 表。DLF Iceberg Catalog 兼容标准 Iceberg REST 协议，使用 AWS SigV4 签名认证，数据面通过社区标准 S3FileIO 直连 OSS S3 兼容端点。

前提条件

运行环境需使用 JDK 17 及以上版本（Iceberg 1.11.0 要求）。
通过 Maven 引入以下 Apache Iceberg 社区开源依赖（版本 1.11.0），所有依赖均可从 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>

说明

如需在本地通过 Java API 直接读写 Parquet 数据（而非通过计算引擎），需额外引入 iceberg-data、iceberg-parquet、iceberg-orc、parquet-hadoop 及 hadoop-common 依赖。

使用 Java API 连接

通过 Iceberg 社区标准 RESTCatalog 类连接 DLF Iceberg REST Catalog，初始化时传入以下配置参数：

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

配置项说明

配置项	说明	示例
`uri`	DLF Iceberg REST 服务地址（VPC 内网），格式为 `http://<region>-vpc.dlf.aliyuncs.com/iceberg`。各 Region 的服务接入点请参见服务接入点与公网访问。	`http://cn-hangzhou-vpc.dlf.aliyuncs.com/iceberg`
`warehouse`	DLF 数据目录（Catalog）名称。	`iceberg_test`
`rest.auth.type`	认证类型，固定值：`sigv4`。	`sigv4`
`rest.auth.sigv4.delegate-auth-type`	委托认证类型，固定值：`none`。	`none`
`rest.signing-region`	DLF 实例所在的 Region ID。	`cn-hangzhou`
`rest.signing-name`	签名服务名称，固定值：`DlfNext`。	`DlfNext`
`rest.access-key-id`	访问 DLF 的 AccessKey ID。
`rest.secret-access-key`	访问 DLF 的 AccessKey Secret。
`io-impl`	文件 IO 实现类，使用 Iceberg 社区标准 S3FileIO，固定值：`org.apache.iceberg.aws.s3.S3FileIO`。	`org.apache.iceberg.aws.s3.S3FileIO`

开源 Spark 访问

开源 Apache Spark 直接使用 Iceberg 社区 Runtime 即可读写 DLF Iceberg 表。Catalog 通过标准 Iceberg REST 协议通信（AWS SigV4 签名认证），数据面通过社区 S3FileIO 以 AWS S3 协议直连 OSS S3 兼容端点。所有依赖均来自 Maven Central。

启动 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

Spark SQL 读写示例

配置完成后，使用标准 Spark SQL 语法读写 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;

开源 Flink 访问

开源 Apache Flink 使用 Iceberg 社区 Runtime 访问 DLF Iceberg 表。将 iceberg-flink-runtime-1.20-1.11.0.jar 与 iceberg-aws-bundle-1.11.0.jar（均为 Maven Central 社区构件）放入 Flink 的 lib/ 目录，然后在 SQL Client 中注册 Catalog。

注册 Catalog

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

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;