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_IDALIBABA_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:
Crie uma única
UploadSessionpara a partição da tabela de destino.Abra um
RecordWriterdedicado para cada thread chamandouploadSession.openRecordWriter(blockId). O ID do bloco identifica exclusivamente o segmento de dados daquela thread. Duas threads não podem compartilhar o mesmo bloco.Preencha os registros em cada thread e grave-os pelo
RecordWritercorrespondente. Em seguida, feche o writer.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 |
|
|
Nome do projeto do MaxCompute |
|
|
|
Nome da tabela de destino |
|
|
|
Especificação da partição |
|
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)) |
|
|
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