Data Lake Formation:Accéder aux tables DLF Paimon via l'API Java

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

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 {

    // Read parameters from environment variables
    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 {
        // Create catalog
        Catalog catalog = createCatalog();

        // List all databases
        System.out.println("=== Databases ===");
        listDatabases(catalog);

        // List tables in a database (replace with your actual database name)
        String database = "default";
        System.out.println("\n=== Tables in " + database + " ===");
        listTables(catalog, database);

        // Read sample data from a table (replace with your actual table name)
        String tableName = "example_table";
        System.out.println("\n=== Sample data from " + tableName + " ===");
        readSampleRows(catalog, database, tableName);
    }

    // Create a Paimon REST Catalog instance
    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));
    }

    // List all databases
    private static void listDatabases(Catalog catalog) throws Exception {
        List<String> databases = catalog.listDatabases();
        for (String database : databases) {
            System.out.println(database);
        }
    }

    // List all tables in a 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);
        }
    }

    // Load a table and read sample data
    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();
                }
            }
        }
    }
}

Paramètres clés