Java API 访问 DLF Paimon 表

前提条件

• 运行环境为 JDK 11 及以上版本。

• 已安装 Maven 3.6 及以上版本。

• 已获取阿里云 AccessKey ID 和 AccessKey Secret。具体操作，请参见创建AccessKey。

• 已创建 DLF 数据目录，并已创建 Paimon 表。具体操作，请参见数据目录和数据表。

• 如果使用 RAM 用户访问 DLF，需要同时配置 API 级别的 RAM 权限策略和数据级别的 DLF 权限。具体操作，请参见快速配置权限。

• 运行环境需要能够访问 DLF VPC Endpoint 和表数据所在的 OSS 内网地址。

准备工作：添加 Maven 依赖

在 pom.xml 中添加 jindodata Maven 仓库和核心依赖。

<properties>
    <paimon.version>1.4.2</paimon.version>
    <jindo.version>6.10.6</jindo.version>
    <hadoop.version>3.3.6</hadoop.version>
</properties>

<repositories>
    <repository>
        <id>jindodata</id>
        <url>https://jindodata-binary.oss-cn-shanghai.aliyuncs.com/mvn-repo/</url>
    </repository>
</repositories>

<dependencies>
    <dependency>
        <groupId>org.apache.paimon</groupId>
        <artifactId>paimon-core</artifactId>
        <version>${paimon.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.paimon</groupId>
        <artifactId>paimon-format</artifactId>
        <version>${paimon.version}</version>
    </dependency>
    <dependency>
        <groupId>com.aliyun.jindodata</groupId>
        <artifactId>jindo-core</artifactId>
        <version>${jindo.version}</version>
    </dependency>
    <dependency>
        <groupId>com.aliyun.jindodata</groupId>
        <artifactId>jindo-sdk</artifactId>
        <version>${jindo.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-common</artifactId>
        <version>${hadoop.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-hdfs-client</artifactId>
        <version>${hadoop.version}</version>
    </dependency>
</dependencies>

使用示例

以下示例演示如何创建 Paimon REST Catalog 实例，并执行列出数据库、列出表、加载表和读取样例数据等操作。

运行前请通过环境变量设置 AccessKey ID、AccessKey Secret、地域和数据目录名称：

export DLF_ACCESS_KEY_ID=<your-access-key-id>
export DLF_ACCESS_KEY_SECRET=<your-access-key-secret>
export DLF_REGION=cn-hangzhou
export DLF_WAREHOUSE=<your-warehouse-name>

import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.options.Options;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.rest.RESTCatalog;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableRead;
import org.apache.paimon.table.source.TableScan;

import java.util.List;

public class DlfPaimonRestExample {

    // 通过环境变量设置以下参数
    private static final String ACCESS_KEY_ID = System.getenv("DLF_ACCESS_KEY_ID");
    private static final String ACCESS_KEY_SECRET = System.getenv("DLF_ACCESS_KEY_SECRET");
    private static final String REGION = System.getenv("DLF_REGION");
    private static final String WAREHOUSE = System.getenv("DLF_WAREHOUSE");
    private static final String DLF_URI =
            "http://" + REGION + "-vpc.dlf.aliyuncs.com";

    public static void main(String[] args) throws Exception {
        // 创建 Catalog
        Catalog catalog = createCatalog();

        // 列出所有数据库
        System.out.println("=== 数据库列表 ===");
        listDatabases(catalog);

        // 列出指定数据库中的表（请替换为实际数据库名称）
        String database = "default";
        System.out.println("\n=== " + database + " 数据库中的表 ===");
        listTables(catalog, database);

        // 读取指定表的样例数据（请替换为实际表名称）
        String tableName = "example_table";
        System.out.println("\n=== " + tableName + " 表的样例数据 ===");
        readSampleRows(catalog, database, tableName);
    }

    // 创建 Paimon REST Catalog 实例
    private static Catalog createCatalog() {
        Options options = new Options();
        options.set("uri", DLF_URI);
        options.set("warehouse", WAREHOUSE);
        options.set("dlf.region", REGION);
        options.set("dlf.access-key-id", ACCESS_KEY_ID);
        options.set("dlf.access-key-secret", ACCESS_KEY_SECRET);
        options.set("token.provider", "dlf");

        return new RESTCatalog(CatalogContext.create(options));
    }

    // 列出所有数据库
    private static void listDatabases(Catalog catalog) throws Exception {
        List<String> databases = catalog.listDatabases();
        for (String database : databases) {
            System.out.println(database);
        }
    }

    // 列出指定数据库中的所有表
    private static void listTables(Catalog catalog, String database) throws Exception {
        List<String> tables = catalog.listTables(database);
        for (String tableName : tables) {
            System.out.println(tableName);
        }
    }

    // 加载表并读取样例数据
    private static void readSampleRows(Catalog catalog, String database, String tableName)
            throws Exception {
        Identifier tableId = Identifier.create(database, tableName);
        Table table = catalog.getTable(tableId);

        ReadBuilder readBuilder = table.newReadBuilder();
        TableScan scan = readBuilder.newScan();
        TableRead read = readBuilder.newRead();

        List<Split> splits = scan.plan().splits();
        for (Split split : splits) {
            try (RecordReader<InternalRow> reader = read.createReader(split)) {
                RecordReader.RecordIterator<InternalRow> batch;
                while ((batch = reader.readBatch()) != null) {
                    InternalRow row;
                    while ((row = batch.next()) != null) {
                        System.out.println(row);
                    }
                    batch.releaseBatch();
                }
            }
        }
    }
}