MaxCompute用户指南：数据上传下载：批量数据通道SDK示例：多线程上传示例

1. import java.io.IOException;
2. import java.util.ArrayList;
3. import java.util.Date;
4. import java.util.concurrent.Callable;
5. import java.util.concurrent.ExecutorService;
6. import java.util.concurrent.Executors;
8. import com.aliyun.odps.Column;
9. import com.aliyun.odps.Odps;
10. import com.aliyun.odps.PartitionSpec;
11. import com.aliyun.odps.TableSchema;
12. import com.aliyun.odps.account.Account;
13. import com.aliyun.odps.account.AliyunAccount;
14. import com.aliyun.odps.data.Record;
15. import com.aliyun.odps.data.RecordWriter;
16. import com.aliyun.odps.tunnel.TableTunnel;
17. import com.aliyun.odps.tunnel.TunnelException;
18. import com.aliyun.odps.tunnel.TableTunnel.UploadSession;
20. class UploadThread implements Callable<Boolean> {
21.          private long id;
22.          private RecordWriter recordWriter;
23.          private Record record;
24.          private TableSchema tableSchema;
26.          public UploadThread(long id, RecordWriter recordWriter, Record record,
27.                          TableSchema tableSchema) {
28.                  this.id = id;
29.                  this.recordWriter = recordWriter;
30.                  this.record = record;
31.                  this.tableSchema = tableSchema;
32.          }
34.          @Override
35.          public Boolean call() {
37.                  for (int i = 0; i < tableSchema.getColumns().size(); i++) {
38.                          Column column = tableSchema.getColumn(i);
39.                          switch (column.getType()) {
40.                          case BIGINT:
41.                                  record.setBigint(i, 1L);
42.                                  break;
43.                          case BOOLEAN:
44.                                  record.setBoolean(i, true);
45.                                  break;
46.                          case DATETIME:
47.                                  record.setDatetime(i, new Date());
48.                                  break;
49.                          case DOUBLE:
50.                                  record.setDouble(i, 0.0);
51.                                  break;
52.                          case STRING:
53.                                  record.setString(i, "sample");
54.                                  break;
55.                          default:
56.                                  throw new RuntimeException("Unknown column type: "
57.                                                  + column.getType());
58.                          }
59.                  }
61.                  for (int i = 0; i < 10; i++) {
62.                          try {
63.                                  recordWriter.write(record);
64.                          } catch (IOException e) {
65.                                  recordWriter.close();
66.                                  e.printStackTrace();
67.                                  return false;
68.                          }
69.                  }
70.                  recordWriter.close();
71.                  return true;
72.          }
74. }
76. public class UploadThreadSample {
77.          private static String accessId = "<your access id>";
78.          private static String accessKey = "<your access Key>";
80.          private static String odpsUrl = "<http://service.odps.aliyun.com/api>";
83.          private static String project = "<your project>";
84.          private static String table = "<your table name>";
85.          private static String partition = "<your partition spec>";
87.          private static int threadNum = 10;
89.          public static void main(String args[]) {
90.                  Account account = new AliyunAccount(accessId, accessKey);
91.                  Odps odps = new Odps(account);
92.                  odps.setEndpoint(odpsUrl);
93.                  odps.setDefaultProject(project);
94.                  try {
95.                          TableTunnel tunnel = new TableTunnel(odps);
96.                          PartitionSpec partitionSpec = new PartitionSpec(partition);
97.                          UploadSession uploadSession = tunnel.createUploadSession(project,
98.                                          table, partitionSpec);
100.                          System.out.println("Session Status is : "
101.                                          + uploadSession.getStatus().toString());
103.                          ExecutorService pool = Executors.newFixedThreadPool(threadNum);
104.                          ArrayList<Callable<Boolean>> callers = new ArrayList<Callable<Boolean>>();
105.                          for (int i = 0; i < threadNum; i++) {
106.                                  RecordWriter recordWriter = uploadSession.openRecordWriter(i);
107.                                  Record record = uploadSession.newRecord();
108.                                  callers.add(new UploadThread(i, recordWriter, record,
109.                                                  uploadSession.getSchema()));
110.                          }
113.                          pool.invokeAll(callers);
114.                          pool.shutdown();
116.                          Long[] blockList = new Long[threadNum];
117.                          for (int i = 0; i < threadNum; i++)
118.                                  blockList = Long.valueOf(i);
119.                          uploadSession.commit(blockList);
120.                          System.out.println("upload success!");
122.                  } catch (TunnelException e) {
123.                          e.printStackTrace();
124.                  } catch (IOException e) {
125.                          e.printStackTrace();
126.                  } catch (InterruptedException e) {
127.                          e.printStackTrace();
128.                  }
129.          }
130. }

注意：
对于 Tunnel Endpoint，支持指定或者不指定。

• 
  如果指定，按照指定的 Endpoint 路由。
• 
  如果不指定，支持自动路由。