MaxCompute:Java SDK examples

Step 1: Configure credentials and environment settings

Store your AccessKey credentials in environment variables rather than hardcoding them in your source code.

// Read credentials from environment variables
String accessId  = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
String accessKey = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");

// Replace <project> with your MaxCompute project name
// Replace <endpoint> with the VPC endpoint for your region
// Replace <quotaName> with your quota name
String project   = "<project>";
String endpoint  = "<endpoint>";
String quotaName = "<quotaName>";

Account account = new AliyunAccount(accessId, accessKey);
Odps odps = new Odps(account);
odps.setDefaultProject(project);
odps.setEndpoint(endpoint);

Credentials credentials = Credentials.newBuilder()
    .withAccount(odps.getAccount())
    .withAppAccount(odps.getAppAccount())
    .build();

EnvironmentSettings settings = EnvironmentSettings.newBuilder()
    .withCredentials(credentials)
    .withServiceEndpoint(odps.getEndpoint())
    .withQuotaName(quotaName)
    .build();

To find the project name, log in to the MaxCompute console, switch to the target region, and go to Workspace > Projects.

To find the quota name:

• Exclusive DTS resource group: Go to Workspace > Quotas in the MaxCompute console. For details, see Manage quotas for computing resources in the MaxCompute console.

• Storage API: Go to Tenants > Tenant Property and enable Storage API Switch.

Job-level quota authorization is not granted by default to any account, including Alibaba Cloud accounts and roles. Authorize access before using the API. For details, see Authorization.

For endpoint values by region, see Endpoints.

Step 2: Create a read session

Build a TableBatchReadSession to specify the table, columns, partitions, and split strategy for your read job.

// Replace <table-name> with your table name
String tableName = "<table-name>";

TableBatchReadSession scan = new TableReadSessionBuilder()
    .identifier(TableIdentifier.of(project, tableName))
    .withSettings(settings)
    .withSplitOptions(SplitOptions.newBuilder()
        .SplitByByteSize(256 * 1024L * 1024L)   // 256 MB per split
        .withCrossPartition(false)
        .build())
    .requiredDataColumns(Arrays.asList("timestamp"))
    .requiredPartitionColumns(Arrays.asList("pt1"))
    .buildBatchReadSession();

In scenarios with large data volumes, network latency, or instability, the creation of data read sessions may take a long time, leading to an automatic switch to an asynchronous process for session creation.

Step 3: Iterate splits and read data

Get the InputSplitAssigner from the session, iterate over each split, and read data using an Arrow reader.

InputSplitAssigner assigner = scan.getInputSplitAssigner();

for (InputSplit split : assigner.getAllSplits()) {
    SplitReader<VectorSchemaRoot> reader = scan.createArrowReader(
        split,
        ReaderOptions.newBuilder()
            .withSettings(settings)
            .withCompressionCodec(CompressionCodec.ZSTD)
            .withReuseBatch(true)
            .build()
    );

    while (reader.hasNext()) {
        VectorSchemaRoot batch = reader.get();
        // batch.getRowCount() returns the number of rows in this batch
        // batch.contentToTSVString() prints the batch as tab-separated values
        System.out.println(batch.contentToTSVString());
    }
    reader.close();
}

When transmitting large amounts of Arrow data, enabling ZSTD or LZ4 compression reduces transfer time caused by network bandwidth limits. If no compression codec is specified, data is sent uncompressed.

Read specific columns (column pruning)

Use requiredDataColumns to read only the columns you need. The returned data preserves the column order you specify.

TableBatchReadSession scan = new TableReadSessionBuilder()
    .identifier(TableIdentifier.of(project, tableName))
    .withSettings(settings)
    .requiredDataColumns(Arrays.asList("user_id", "event_time", "action"))
    .buildBatchReadSession();

If requiredDataColumns is empty, all partition data is returned.

Read specific partitions (partition pruning)

Use requiredPartitions to limit the read to specific partitions. If requiredPartitions is empty, all partitions are read.

Push down filter predicates

Use withFilterPredicate to push filter conditions to the server, reducing the amount of data transferred.

// Filter: c1 > 20000 AND c2 < 100000
BinaryPredicate c1 = new BinaryPredicate(
    BinaryPredicate.Operator.GREATER_THAN,
    Attribute.of("c1"),
    Constant.of(20000)
);
BinaryPredicate c2 = new BinaryPredicate(
    BinaryPredicate.Operator.LESS_THAN,
    Attribute.of("c2"),
    Constant.of(100000)
);
CompoundPredicate predicate = new CompoundPredicate(
    CompoundPredicate.Operator.AND,
    ImmutableList.of(c1, c2)
);

TableBatchReadSession scan = new TableReadSessionBuilder()
    .identifier(TableIdentifier.of(project, tableName))
    .withSettings(settings)
    .withFilterPredicate(predicate)
    .buildBatchReadSession();

Read by row offset

Use SplitByRowOffset when you need precise control over which rows each split covers — for example, to distribute work evenly across parallel readers.

TableBatchReadSession scan = new TableReadSessionBuilder()
    .identifier(TableIdentifier.of(project, tableName))
    .withSettings(settings)
    .withSplitOptions(SplitOptions.newBuilder()
        .SplitByRowOffset()
        .build())
    .buildBatchReadSession();

InputSplitAssigner assigner = scan.getInputSplitAssigner();
long totalRows       = assigner.getTotalRowCount();
long recordsPerSplit = 10000;

for (long offset = 0; offset < totalRows; offset += recordsPerSplit) {
    long count = Math.min(recordsPerSplit, totalRows - offset);
    InputSplit split = assigner.getSplitByRowOffset(offset, count);
    // Create a reader and process the split
}

TableReadSessionBuilder

Builds a TableBatchReadSession. For the full Javadoc, see odps-sdk-table-api Javadoc.

public class SplitOptions {
    public static SplitOptions.Builder newBuilder();

    public static class Builder {
        public Builder SplitByByteSize(long splitByteSize);
        public Builder SplitByRowOffset();
        public Builder withCrossPartition(boolean crossPartition);
        public Builder withMaxFileNum(int splitMaxFileNum);
        public SplitOptions build();
    }
}

// Split by byte size
SplitOptions bySize = SplitOptions.newBuilder()
    .SplitByByteSize(256 * 1024L * 1024L)
    .build();

// Split by row offset
SplitOptions byRow = SplitOptions.newBuilder()
    .SplitByRowOffset()
    .build();

// Split by row offset, max 1 file per split
SplitOptions byRowOneFile = SplitOptions.newBuilder()
    .SplitByRowOffset()
    .withMaxFileNum(1)
    .build();

public class ArrowOptions {
    public static Builder newBuilder();

    public static class Builder {
        public Builder withTimestampUnit(TimestampUnit unit);
        public Builder withDatetimeUnit(TimestampUnit unit);
        public ArrowOptions build();
    }

    public enum TimestampUnit { SECOND, MILLI, MICRO, NANO }
}

ArrowOptions options = ArrowOptions.newBuilder()
    .withDatetimeUnit(ArrowOptions.TimestampUnit.MILLI)
    .withTimestampUnit(ArrowOptions.TimestampUnit.NANO)
    .build();

// c1 > 20000 AND c2 < 100000
BinaryPredicate c1 = new BinaryPredicate(BinaryPredicate.Operator.GREATER_THAN,  Attribute.of("c1"), Constant.of(20000));
BinaryPredicate c2 = new BinaryPredicate(BinaryPredicate.Operator.LESS_THAN,     Attribute.of("c2"), Constant.of(100000));
CompoundPredicate predicate = new CompoundPredicate(CompoundPredicate.Operator.AND, ImmutableList.of(c1, c2));

// c1 IS NOT NULL
Predicate notNull = new UnaryPredicate(UnaryPredicate.Operator.NOT_NULL, Attribute.of("c1"));

// c1 IN (1, 10001)
Predicate inList = new InPredicate(InPredicate.Operator.IN, Attribute.of("c1"),
    ImmutableList.of(Constant.of(1), Constant.of(10001)));

public class EnvironmentSettings {
    public static Builder newBuilder();

    public static class Builder {
        public Builder withDefaultProject(String projectName);
        public Builder withDefaultSchema(String schema);
        public Builder withServiceEndpoint(String endpoint);
        public Builder withTunnelEndpoint(String tunnelEndpoint);
        public Builder withQuotaName(String quotaName);
        public Builder withCredentials(Credentials credentials);
        public Builder withRestOptions(RestOptions restOptions);
        public EnvironmentSettings build();
    }
}

TableBatchReadSession

TableBatchReadSession represents an active read session. Retrieve it from TableReadSessionBuilder.buildBatchReadSession().

public interface TableBatchReadSession {
    String getId();
    TableIdentifier getTableIdentifier();
    SessionStatus getStatus();
    DataSchema readSchema();
    InputSplitAssigner getInputSplitAssigner() throws IOException;
    SplitReader<ArrayRecord>       createRecordReader(InputSplit split, ReaderOptions options) throws IOException;
    SplitReader<VectorSchemaRoot>  createArrowReader(InputSplit split, ReaderOptions options)  throws IOException;
}

// Distribute by byte-size splits
InputSplitAssigner assigner = scan.getInputSplitAssigner();
int splitCount = assigner.getSplitsCount();
for (int i = 0; i < splitCount; i++) {
    InputSplit split = assigner.getSplit(i);
    // process split
}

// Distribute by row offset
long totalRows       = assigner.getTotalRowCount();
long recordsPerSplit = 10000;
for (long offset = 0; offset < totalRows; offset += recordsPerSplit) {
    long count = Math.min(recordsPerSplit, totalRows - offset);
    InputSplit split = assigner.getSplitByRowOffset(offset, count);
    // process split
}

public class ReaderOptions {
    public static ReaderOptions.Builder newBuilder();

    public static class Builder {
        public Builder withMaxBatchRowCount(int maxBatchRowCount);
        public Builder withMaxBatchRawSize(long batchRawSize);
        public Builder withCompressionCodec(CompressionCodec codec);
        public Builder withBufferAllocator(BufferAllocator allocator);
        public Builder withReuseBatch(boolean reuseBatch);
        public Builder withSettings(EnvironmentSettings settings);
        public ReaderOptions build();
    }
}

SplitReader

SplitReader<T> reads data from a single split. Use SplitReader<ArrayRecord> for record format or SplitReader<VectorSchemaRoot> for Arrow format.

public interface SplitReader<T> {
    boolean hasNext() throws IOException;
    T get();
    Metrics currentMetricsValues();
    void close() throws IOException;
}