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(); } } }