MaxCompute快速入门：导入数据

MaxCompute 提供多种数据导入导出方式，如下所示：

• 
  直接在客户端使用 Tunnel 命令。
• 
  通过 Tunnel 提供的 SDK 自行编写 Java 工具。
• 
  通过 Flume 及 Fluentd 插件方式导入。
• 
  通过大数据开发套件对数据导入和导出，详情请参见 数据集成概述。导出数据请参见 Tunnel 命令操作 中 Download 的相关命令。

Tunnel 命令导入数据



准备数据


假设您已准备本地文件 wc_example.txt，本地存放路径为 D:\odps\odps\bin，内容如下：  

1. I LOVE CHINA!
2. MY NAME IS MAGGIE.I LIVE IN HANGZHOU!I LIKE PLAYING BASKETBALL!

创建 MaxCompute 表


您需要把上面的数据导入到 MaxCompute 的一张表中，所以需要创建 MaxCompute 表：

1. CREATE TABLE wc_in (word string);

执行 tunnel 命令


输入表创建成功后，可以在 MaxCompute 客户端输入 Tunnel 命令进行数据的导入，如下所示：  

1. tunnel upload D:\odps\odps\bin\wc_example.txt wc_in;

执行成功后，查看表 wc_in 的记录，如下所示：

注意：

• 
  有关 Tunnel 命令的更多详细介绍，例如：如何将数据导入分区表等，请参见 Tunnel 操作。
• 
  当表中含有多个列时，可以通过 -fd 参数指定列分隔符。

Tunnel SDK


下文将通过场景示例，为您介绍如何利用 Tunnel SDK 上传数据。

场景描述


上传数据到 MaxCompute，其中，项目空间为 odps_public_dev，表名为 tunnel_sample_test，分区为 pt=20150801，dt=hangzhou。

操作步骤

1. 
   创建表，添加分区，SQL 语句如下所示：CREATE TABLE IF NOT EXISTS tunnel_sample_test(
2. id STRING,
3. name STRING)
4. PARTITIONED BY (pt STRING, dt STRING); --创建表
5. ALTER TABLE tunnel_sample_test
6. ADD IF NOT EXISTS PARTITION (pt='20150801',dt='hangzhou'); --添加分区

创建 UploadSample 的工程目录结构，如下所示：  

1. |---pom.xml
2. |---src
3. |---main
4.     |---java  
5.        |---com  
6.            |---aliyun
7.                |---odps  
8.                    |---tunnel  
9.                        |---example  
10.                            |---UploadSample.java

目录说明：

• 
  pom.xml：maven 工程文件。
• 
  UploadSample：Tunnel 源文件。

编写 UploadSample 程序。代码如下所示：

1. package com.aliyun.odps.tunnel.example;
2. import java.io.IOException;
3. import java.util.Date;
5. import com.aliyun.odps.Column;
6. import com.aliyun.odps.Odps;
7. import com.aliyun.odps.PartitionSpec;
8. import com.aliyun.odps.TableSchema;
9. import com.aliyun.odps.account.Account;
10. import com.aliyun.odps.account.AliyunAccount;
11. import com.aliyun.odps.data.Record;
12. import com.aliyun.odps.data.RecordWriter;
13. import com.aliyun.odps.tunnel.TableTunnel;
14. import com.aliyun.odps.tunnel.TunnelException;
15. import com.aliyun.odps.tunnel.TableTunnel.UploadSession;
17. public class UploadSample {
18.       private static String accessId = "####";
19.       private static String accessKey = "####";
20.       private static String tunnelUrl = "http://dt.odps.aliyun.com";
22.       private static String odpsUrl = "http://service.odps.aliyun.com/api";
24.       private static String project = "odps_public_dev";
25.       private static String table = "tunnel_sample_test";
26.       private static String partition = "pt=20150801,dt=hangzhou";
28.       public static void main(String args[]) {
29.               Account account = new AliyunAccount(accessId, accessKey);
30.               Odps odps = new Odps(account);
31.               odps.setEndpoint(odpsUrl);
32.               odps.setDefaultProject(project);
33.               try {
34.                       TableTunnel tunnel = new TableTunnel(odps);
35.                       tunnel.setEndpoint(tunnelUrl);
36.                       PartitionSpec partitionSpec = new PartitionSpec(partition);
37.                       UploadSession uploadSession = tunnel.createUploadSession(project,
38.                                       table, partitionSpec);
40.                       System.out.println("Session Status is : "
41.                                       + uploadSession.getStatus().toString());
43.                       TableSchema schema = uploadSession.getSchema();
44.                       RecordWriter recordWriter = uploadSession.openRecordWriter(0);
45.                       Record record = uploadSession.newRecord();
46.                       for (int i = 0; i < schema.getColumns().size(); i++) {
47.                               Column column = schema.getColumn(i);
48.                               switch (column.getType()) {
49.                               case BIGINT:
50.                                       record.setBigint(i, 1L);
51.                                       break;
52.                               case BOOLEAN:
53.                                       record.setBoolean(i, true);
54.                                       break;
55.                               case DATETIME:
56.                                       record.setDatetime(i, new Date());
57.                                       break;
58.                               case DOUBLE:
59.                                       record.setDouble(i, 0.0);
60.                                       break;
61.                               case STRING:
62.                                       record.setString(i, "sample");
63.                                       break;
64.                               default:
65.                                       throw new RuntimeException("Unknown column type: "
66.                                                       + column.getType());
67.                               }
68.                       }
69.                       for (int i = 0; i < 10; i++) {
70.                               recordWriter.write(record);
71.                       }
72.                       recordWriter.close();
73.                       uploadSession.commit(new Long[]{0L});
74.                       System.out.println("upload success!");
76.               } catch (TunnelException e) {
77.                       e.printStackTrace();
78.               } catch (IOException e) {
79.                       e.printStackTrace();
80.               }
81.       }
82. }

注意：
这里省略了 accessId 和 accesskey 的配置，实际运行时请换上您自己的 accessId 以及 accessKey。

配置 pom.xml 文件。如下所示：

1. <?xml version="1.0" encoding="UTF-8"?>
2. <project xmlns="http://maven.apache.org/POM/4.0.0"
3. xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4. xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
6. <modelVersion>4.0.0</modelVersion>
7. <groupId>com.aliyun.odps.tunnel.example</groupId>
8. <artifactId>UploadSample</artifactId>
9. <version>1.0-SNAPSHOT</version>
11. <dependencies>
12.   <dependency>
13.     <groupId>com.aliyun.odps</groupId>
14.     <artifactId>odps-sdk-core</artifactId>
15.     <version>0.20.7-public</version>
16.   </dependency>
17. </dependencies>
19. <repositories>
20.   <repository>
21.   <id>alibaba</id>
22.   <name>alibaba Repository</name>
23.   <url>http://mvnrepo.alibaba-inc.com/nexus/content/groups/public/</url>
24.   </repository>
25. </repositories>
27. </project>

编译与运行。
编译 UploadSample 工程，如下所示：

1. mvn package

运行 UploadSample 程序，这里使用 eclipse 导入 maven project：

1. 
   右击 java 工程 并单击 Import->Maven->Existing Maven Projects 设置如下：
   
2. 
   右击 UploadSample.java 并单击 Run As->Run Configurations，如下所示：
   
3. 
   单击 Run 运行成功，控制台显示如下：  Session Status is : NORMAL
4. upload success!

查看运行结果。
您在客户端输入如下语句，即可查看运行结果。  

1. select * from tunnel_sample_test;

显示结果如下：  

1. +----+------+----+----+
2. | id | name | pt | dt |
3. +----+------+----+----+
4. | sample | sample | 20150801 | hangzhou |
5. | sample | sample | 20150801 | hangzhou |
6. | sample | sample | 20150801 | hangzhou |
7. | sample | sample | 20150801 | hangzhou |
8. | sample | sample | 20150801 | hangzhou |
9. | sample | sample | 20150801 | hangzhou |
10. | sample | sample | 20150801 | hangzhou |
11. | sample | sample | 20150801 | hangzhou |
12. | sample | sample | 20150801 | hangzhou |
13. | sample | sample | 20150801 | hangzhou |
14. +----+------+----+----+

注意：

• 
  Tunnel 作为 MaxCompute 中一个独立的服务，有专属的访问端口提供给大家。当您在阿里云内网环境中，使用 Tunnel 内网连接下载数据时，MaxCompute 不会将该操作产生的流量计入计费。此外内网地址仅对上海域的云产品有效。
• 
  MaxCompute 阿里云内网地址：http://odps-ext.aliyun-inc.com/api 。
• 
  MaxCompute 公网地址：http://service.odps.aliyun.com/api 。

其他导入方式


除了通过客户端及 Tunnel Java SDK 导入数据，阿里云数加数据集成、开源的Sqoop、Fluentd、Flume、LogStash 等工具都可以进行数据导入到 MaxCompute，具体介绍请参见 数据上传下载-工具介绍。