Data upload by using BufferedWriter in multi-threaded mode-MaxCompute(MaxCompute)-阿里云帮助中心

When uploading large volumes of data, a single-threaded writer can become a bottleneck. TunnelBufferedWriter lets multiple threads write to the same MaxCompute table concurrently from a shared UploadSession, increasing throughput without requiring separate sessions per thread.

How it works

Each thread opens its own TunnelBufferedWriter instance from a shared UploadSession. Threads write records independently and close their writers when done. After all threads finish, call commit() on the session to finalize the upload.

Important

TunnelBufferedWriter is not thread-safe. Each thread must create and manage its own writer instance by calling openBufferedWriter() independently. Sharing a single writer across threads causes data corruption or errors.

Prerequisites

Before you begin, ensure that you have:

Initialized a TableTunnel instance with valid MaxCompute and Tunnel credentials
A projectName and tableName for the target table

Upload data in multi-threaded mode

The following example creates two threads that each upload 1,200 records to a MaxCompute table concurrently.

// Each upload thread opens its own TunnelBufferedWriter and writes records independently.
class UploadThread extends Thread {
  private UploadSession session;
  private static int RECORD_COUNT = 1200;

  public UploadThread(UploadSession session) {
    this.session = session;
  }

  @Override
  public void run() {
    // Each thread must open its own writer — TunnelBufferedWriter is not thread-safe.
    RecordWriter writer = session.openBufferedWriter();
    Record r = session.newRecord();
    try {
      for (int i = 0; i < RECORD_COUNT; i++) {
        r.setBigint(0, i);
        writer.write(r);
      }
    } finally {
      // Always close the writer to flush buffered data before the session commits.
      writer.close();
    }
  }
}

public class Example {
  public static void main(String[] args) throws Exception {
    // Initialize MaxCompute and Tunnel, then create an upload session.
    TableTunnel.UploadSession uploadSession =
        tunnel.createUploadSession(projectName, tableName);

    UploadThread t1 = new UploadThread(uploadSession);
    UploadThread t2 = new UploadThread(uploadSession);

    t1.start();
    t2.start();

    // Wait for both threads to finish before committing.
    t1.join();
    t2.join();

    // Commit the session to finalize the upload.
    uploadSession.commit();
  }
}

Replace the following placeholders with actual values:

Placeholder	Description	Example
`projectName`	The name of the MaxCompute project	`my_project`
`tableName`	The name of the target table	`my_table`

Concurrency and session management

Rule	Detail
One writer per thread	Each thread must call `openBufferedWriter()` independently. Sharing a writer instance across threads causes data corruption or errors.
Close before commit	Call `writer.close()` in each thread before calling `uploadSession.commit()`. Data in an unclosed writer may not be flushed to the server.
Commit after all threads finish	Call `uploadSession.commit()` only after joining all upload threads. Committing while threads are still writing causes the upload to fail.
Session reuse across threads	A single `UploadSession` can be shared across threads. Each thread opens its own `TunnelBufferedWriter` from the shared session.

Error handling

Handle write errors and commit failures explicitly in production code.

Write errors: Wrap writer.write(r) in a try-catch block to handle TunnelException or IOException. If a thread fails mid-write, call writer.close() in a finally block to release resources, then allow the exception to propagate so that the main thread can abort the session instead of committing partial data.

Commit failures: If uploadSession.commit() throws an exception, the upload is not finalized. You can retry the commit on transient errors. If the commit continues to fail, create a new UploadSession and retry the full upload.