Multithreading downloads

How it works

1. Create a DownloadSession to get the total record count for the target table or partition.

2. Divide the records into equal-sized ranges — one range per thread.

3. Each thread opens its own RecordReader for its assigned range and reads records concurrently.

4. Sum the per-thread record counts to verify all records were downloaded.

All threads share the same DownloadSession, but each thread has an independent RecordReader to avoid contention.

Prerequisites

Before you begin, make sure you have:

• A MaxCompute project with an accessible table

• An AccessKey ID and AccessKey secret (stored as environment variables, not hard-coded)

• The MaxCompute Java SDK added to your project dependencies

• The Tunnel endpoint for your target region (see Endpoints)

Download data with multiple threads

The following example creates a 10-thread download session for a partitioned table. Each thread reads an independent slice of the data using RecordReader.

import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import com.aliyun.odps.Column;
import com.aliyun.odps.Odps;
import com.aliyun.odps.PartitionSpec;
import com.aliyun.odps.TableSchema;
import com.aliyun.odps.account.Account;
import com.aliyun.odps.account.AliyunAccount;
import com.aliyun.odps.data.Record;
import com.aliyun.odps.data.RecordReader;
import com.aliyun.odps.tunnel.TableTunnel;
import com.aliyun.odps.tunnel.TableTunnel.DownloadSession;
import com.aliyun.odps.tunnel.TunnelException;

// Worker thread: reads a contiguous range of records from a DownloadSession
class DownloadThread implements Callable<Long> {
    private long id;
    private RecordReader recordReader;
    private TableSchema tableSchema;

    public DownloadThread(int id, RecordReader recordReader, TableSchema tableSchema) {
        this.id = id;
        this.recordReader = recordReader;
        this.tableSchema = tableSchema;
    }

    @Override
    public Long call() {
        Long recordNum = 0L;
        try {
            Record record;
            while ((record = recordReader.read()) != null) {
                recordNum++;
                System.out.print("Thread " + id + "\t");
                consumeRecord(record, tableSchema);
            }
            recordReader.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return recordNum;
    }

    // Reads each column value and prints it tab-separated
    private static void consumeRecord(Record record, TableSchema schema) {
        for (int i = 0; i < schema.getColumns().size(); i++) {
            Column column = schema.getColumn(i);
            String colValue = null;
            switch (column.getType()) {
                case BIGINT: {
                    Long v = record.getBigint(i);
                    colValue = v == null ? null : v.toString();
                    break;
                }
                case BOOLEAN: {
                    Boolean v = record.getBoolean(i);
                    colValue = v == null ? null : v.toString();
                    break;
                }
                case DATETIME: {
                    Date v = record.getDatetime(i);
                    colValue = v == null ? null : v.toString();
                    break;
                }
                case DOUBLE: {
                    Double v = record.getDouble(i);
                    colValue = v == null ? null : v.toString();
                    break;
                }
                case STRING: {
                    String v = record.getString(i);
                    colValue = v == null ? null : v.toString();
                    break;
                }
                default:
                    throw new RuntimeException("Unknown column type: " + column.getType());
            }
            System.out.print(colValue == null ? "null" : colValue);
            if (i != schema.getColumns().size() - 1) System.out.print("\t");
        }
        System.out.println();
    }
}

public class DownloadThreadSample {
    // Read credentials from environment variables — never hard-code AccessKey pairs
    private static String accessId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
    private static String accessKey = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");

    private static String odpsUrl = "http://service.odps.aliyun.com/api";
    // Tunnel endpoint for the cloud product interconnection network in China (Shanghai)
    // Replace with the endpoint for your target region. Leave blank to use the public endpoint.
    private static String tunnelUrl = "http://dt.cn-shanghai.maxcompute.aliyun-inc.com";

    private static String project = "<your-project>";
    private static String table = "<your-table>";
    private static String partition = "<your-partition-spec>";
    private static int threadNum = 10;

    public static void main(String args[]) {
        Account account = new AliyunAccount(accessId, accessKey);
        Odps odps = new Odps(account);
        odps.setEndpoint(odpsUrl);
        odps.setDefaultProject(project);

        TableTunnel tunnel = new TableTunnel(odps);
        tunnel.setEndpoint(tunnelUrl);

        PartitionSpec partitionSpec = new PartitionSpec(partition);
        DownloadSession downloadSession;

        try {
            // Create a session and get the total record count
            downloadSession = tunnel.createDownloadSession(project, table, partitionSpec);
            System.out.println("Session status: " + downloadSession.getStatus().toString());
            long count = downloadSession.getRecordCount();
            System.out.println("Total records: " + count);

            // Divide records into threadNum equal ranges.
            // Thread i reads records [step*i, step*i + step).
            // The last thread handles any remaining records to avoid data loss.
            ExecutorService pool = Executors.newFixedThreadPool(threadNum);
            ArrayList<Callable<Long>> callers = new ArrayList<Callable<Long>>();
            long step = count / threadNum;

            for (int i = 0; i < threadNum - 1; i++) {
                RecordReader recordReader = downloadSession.openRecordReader(step * i, step);
                callers.add(new DownloadThread(i, recordReader, downloadSession.getSchema()));
            }
            // Last thread reads from step*(threadNum-1) to the end of the dataset
            RecordReader recordReader = downloadSession.openRecordReader(
                step * (threadNum - 1),
                count - ((threadNum - 1) * step)
            );
            callers.add(new DownloadThread(threadNum - 1, recordReader, downloadSession.getSchema()));

            // Run all threads and sum the record counts
            Long downloadNum = 0L;
            List<Future<Long>> recordNum = pool.invokeAll(callers);
            for (Future<Long> num : recordNum) downloadNum += num.get();
            System.out.println("Records downloaded: " + downloadNum);
            pool.shutdown();

        } catch (TunnelException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }
}

Replace the following placeholders before running:

Configure the Tunnel endpoint

The tunnelUrl parameter controls which network path is used for data transfer.

The example uses the cloud product interconnection network Tunnel endpoint for China (Shanghai). For Tunnel endpoints in other regions, see Endpoints.

Best practices

Choose the right thread count. Start with 10 threads (the example default) and tune based on your network bandwidth and table size. Adding threads beyond your available bandwidth or CPU cores yields diminishing returns.

Handle the last-thread remainder. When count is not evenly divisible by threadNum, the last thread reads the remaining records (count - (threadNum - 1) * step). The example already implements this — do not replace it with a fixed step size, or you will lose data.

Verify record counts after download. Always compare the total downloaded record count against downloadSession.getRecordCount(). A mismatch indicates a thread failed silently. The example catch blocks only print stack traces; in production, log the error and decide whether to retry or abort the session.

Do not reuse a failed session. A DownloadSession that encounters a TunnelException is in an undefined state. Create a new session rather than retrying with the same instance.

Omit `partitionSpec` for non-partitioned tables. Call tunnel.createDownloadSession(project, table) instead of the three-argument form. Passing a PartitionSpec to a non-partitioned table causes an error.

Next steps

• Endpoints — Tunnel endpoints for all regions and network types