import java.io.BufferedReader; import java.io.IOException; import java.io.InputStreamReader; import java.util.ArrayList; 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.TableTunnel.UploadSession; import com.aliyun.odps.tunnel.TunnelException; import com.aliyun.oss.OSSClient; import com.aliyun.oss.model.GetObjectRequest; import com.aliyun.oss.model.OSSObject; class UploadThread implements Callable<Boolean> { private long id; private TableSchema schema = null; private RecordWriter recordWriter = null; private Record record = null; private BufferedReader reader = null; public UploadThread(long id, RecordWriter recordWriter, Record record,                          TableSchema schema,BufferedReader reader) { this.id = id; this.recordWriter = recordWriter; this.record = record; this.schema = schema; this.reader = reader; } public Boolean call() throws Exception { while (true) {                    String line = reader.readLine(); if (line == null) break; if(id == 0){ //第一行是字段名，忽略掉                           id++; continue; }                  System.out.println(line);                   String[] s = line.split(","); for (int i = 0; i < schema.getColumns().size(); i++) {                       Column column = schema.getColumn(i); switch (column.getType()) { case BIGINT:                              record.setBigint(i, Long.valueOf(s[i])); break; //               case BOOLEAN: //                       record.setBoolean(i, str); //                       break; //               case DATETIME: //                       record.setDatetime(i, str); //                       break; case DOUBLE:                                record.setDouble(i, Double.valueOf(s[i])); break; case STRING:                              record.setString(i,s[i]); break; default: throw new RuntimeException("Unknown column type: " + column.getType()); } }                    recordWriter.write(record); }               recordWriter.close(); return true; } } public class OSSToODPS_UploadThread { private static String accessKeyId = "UQV2yoSSWNgquhhe"; private static String accessKeySecret = "bG8xSLwhmKYRmtBoE3HbhOBYXvknG6"; private static String endpoint = "http://oss-cn-hangzhou.aliyuncs.com"; private static String bucketName = "qf-test"; private static String key = "data_test/movies.csv"; private static String tunnelUrl = "http://dt.odps.aliyun.com"; private static String odpsUrl = "http://service.odps.aliyun.com/api"; private static String project = "dtstack_dev"; private static String table = "movies_odps2"; //private static String partition = "ds=20160612"; private static int threadNum = 10; public static void main(String args[]) { /*                   * Constructs a client instance with your account for accessing OSS                  */                 OSSClient client = new OSSClient(endpoint, accessKeyId, accessKeySecret);                  System.out.println("Downloading an object");                  OSSObject object = client.getObject(new GetObjectRequest(bucketName, key));                BufferedReader reader = new BufferedReader(new InputStreamReader(object.getObjectContent()));                                    Account account = new AliyunAccount(accessKeyId, accessKeySecret);                  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);                          UploadSession uploadSession = tunnel.createUploadSession(project,table); //                       UploadSession uploadSession = tunnel.createUploadSession(project, //                                 table, partitionSpec);  //分区                           System.out.println("Session Status is : " + uploadSession.getStatus().toString());                           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(),reader)); }                          pool.invokeAll(callers);                          pool.shutdown();                           Long[] blockList = new Long[threadNum]; for (int i = 0; i < threadNum; i++)                                  blockList[i] = Long.valueOf(i);                          uploadSession.commit(blockList);                          reader.close();                          System.out.println("upload success!"); } catch (TunnelException e) {                          e.printStackTrace(); } catch (IOException e) {                          e.printStackTrace(); } catch (InterruptedException e) {                          e.printStackTrace(); } } }