Todos os produtos
Search
Central de documentação

MaxCompute:Upload data in multithreaded mode

Última atualização: Jun 26, 2026

Faça upload de dados para o MaxCompute em paralelo usando a interface TableTunnel com o ExecutorService do Java. Cada thread possui um bloco exclusivo e um RecordWriter próprio. Não compartilhe blocos entre threads.

Pré-requisitos

Antes de começar, verifique se você tem:

  • Um projeto do MaxCompute com uma tabela particionada para gravação

  • O SDK Java do MaxCompute adicionado às dependências do seu projeto

  • Um AccessKey ID e um AccessKey secret armazenados como variáveis de ambiente:

    • ALIBABA_CLOUD_ACCESS_KEY_ID

    • ALIBABA_CLOUD_ACCESS_KEY_SECRET

  • (Recomendado) Um usuário RAM com as permissões mínimas necessárias, em vez da conta raiz da Alibaba Cloud. As credenciais da conta raiz apresentam alto risco porque o par de AccessKey tem permissões em todas as operações de API. Crie um usuário RAM no console do Resource Access Management (RAM).

Como funciona

O fluxo de upload envolve quatro etapas:

  1. Crie uma única UploadSession para a partição da tabela de destino.

  2. Abra um RecordWriter dedicado para cada thread chamando uploadSession.openRecordWriter(blockId). O ID do bloco identifica exclusivamente o segmento de dados daquela thread. Duas threads não podem compartilhar o mesmo bloco.

  3. Preencha os registros em cada thread e grave-os pelo RecordWriter correspondente. Em seguida, feche o writer.

  4. Após a conclusão de todas as threads, confirme a sessão de upload com a lista completa de IDs de bloco.

Upload de dados com múltiplas threads

O exemplo a seguir cria 10 threads de upload. Cada uma grava 10 registros em seu próprio bloco.

import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
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.RecordWriter;
import com.aliyun.odps.tunnel.TableTunnel;
import com.aliyun.odps.tunnel.TunnelException;
import com.aliyun.odps.tunnel.TableTunnel.UploadSession;

// Each thread owns one block (RecordWriter) and writes 10 records to it.
class UploadThread implements Callable<Boolean> {
    private long id;
    private RecordWriter recordWriter;
    private Record record;
    private TableSchema tableSchema;

    public UploadThread(long id, RecordWriter recordWriter, Record record,
                    TableSchema tableSchema) {
        this.id = id;
        this.recordWriter = recordWriter;
        this.record = record;
        this.tableSchema = tableSchema;
    }

    @Override
    public Boolean call() {
        // Populate each column based on its data type
        for (int i = 0; i < tableSchema.getColumns().size(); i++) {
            Column column = tableSchema.getColumn(i);
            switch (column.getType()) {
            case BIGINT:
                record.setBigint(i, 1L);
                break;
            case BOOLEAN:
                record.setBoolean(i, true);
                break;
            case DATETIME:
                record.setDatetime(i, new Date());
                break;
            case DOUBLE:
                record.setDouble(i, 0.0);
                break;
            case STRING:
                record.setString(i, "sample");
                break;
            default:
                throw new RuntimeException("Unknown column type: "
                                + column.getType());
            }
        }

        boolean success = true;
        try {
            for (int i = 0; i < 10; i++) {
                recordWriter.write(record);
            }
        } catch (IOException e) {
            success = false;
            e.printStackTrace();
        } finally {
            recordWriter.close();
        }
        return success;
    }
}

public class UploadThreadSample {
    // Load credentials from environment variables—never hardcode them in source code.
    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>";

    // Set tunnelUrl to a specific Tunnel endpoint, or leave it blank to use the public endpoint.
    // The example below uses the classic network Tunnel endpoint in the China (Shanghai) region.
    private static String tunnelUrl = "<http://dt.cn-shanghai.maxcompute.aliyun-inc.com>";

    private static String project = "<your project>";
    private static String table = "<your table name>";
    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);

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

            PartitionSpec partitionSpec = new PartitionSpec(partition);

            // Create a single upload session for the target partition
            UploadSession uploadSession = tunnel.createUploadSession(project,
                            table, partitionSpec);
            System.out.println("Session Status is : " + uploadSession.getStatus().toString());

            // Assign one RecordWriter (block) per thread
            ExecutorService pool = Executors.newFixedThreadPool(threadNum);
            ArrayList<Callable<Boolean>> callers = new ArrayList<Callable<Boolean>>();
            for (int i = 0; i < threadNum; i++) {
                RecordWriter recordWriter = uploadSession.openRecordWriter(i);
                Record record = uploadSession.newRecord();
                callers.add(new UploadThread(i, recordWriter, record,
                                uploadSession.getSchema()));
            }

            // Run all threads and wait for them to finish
            pool.invokeAll(callers);
            pool.shutdown();

            // Commit all blocks
            Long[] blockList = new Long[threadNum];
            for (int i = 0; i < threadNum; i++)
                blockList[i] = Long.valueOf(i);
            uploadSession.commit(blockList);
            System.out.println("upload success!");

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

Substitua os seguintes espaços reservados antes de executar o código:

Espaço reservado

Descrição

Exemplo

<your project>

Nome do projeto do MaxCompute

my_project

<your table name>

Nome da tabela de destino

my_table

<your partition spec>

Especificação da partição

ds=20240101

Configuração do endpoint do Tunnel

Defina o tunnelUrl conforme o tipo de rede:

Cenário

tunnelUrl

Deixar em branco

Endpoint público usado por padrão

Rede interna (rede clássica, China (Xangai))

http://dt.cn-shanghai.maxcompute.aliyun-inc.com

Outras regiões ou tipos de rede

Consulte Endpoints

Verificação do upload

Depois que o programa exibir upload success!, verifique os dados no MaxCompute executando:

SELECT COUNT(*) FROM <your table name> WHERE <your partition spec>;

A contagem deve corresponder ao total de registros gravados em todas as threads. Neste exemplo: 10 threads x 10 registros = 100 registros.

Segurança

Armazene as credenciais do AccessKey em variáveis de ambiente, nunca no código-fonte. Para cargas de trabalho de produção, use um usuário RAM com as permissões mínimas necessárias em vez das credenciais da conta raiz. Para criar um usuário RAM, acesse o console do Resource Access Management (RAM).

Próximos passos

  • Endpoints — encontre o endpoint do Tunnel para sua região e tipo de rede