Tunnel SDK

關於如何利用tunnel SDK進行上傳數據，下麵也將通過場景介紹。場景描述：上傳數據到ODPS，其中，項目空間為”odps_public_dev”，表名為”tunnel_sample_test”，分區為”pt=20150801,dt=hangzhou”。

1. 創建表，添加分區：

   CREATE TABLE IF NOT EXISTS tunnel_sample_test(
    id STRING,
    name STRING)
   PARTITIONED BY (pt STRING, dt STRING); --創建表
   ALTER TABLE tunnel_sample_test
    ADD IF NOT EXISTS PARTITION (pt='20150801',dt='hangzhou'); --添加分區

2. 創建UploadSample的工程目錄結構，如下：

   |---pom.xml
   |---src 
     |---main
        |---java  
           |---com  
               |---aliyun
                   |---odps  
                       |---tunnel  
                           |---example  
                               |---UploadSample.java

   UploadSample : tunnel源文件pom.xml : maven工程文件

3. 編寫UploadSample程序，程序如下：

   package com.aliyun.odps.tunnel.example;
   import java.io.IOException;
   import java.util.Date;
   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;
   public class UploadSample {
         private static String accessId = "####";
         private static String accessKey = "####";
         private static String tunnelUrl = "https://dt-corp.odps.aliyun-inc.com";
         private static String odpsUrl = "https://service-corp.odps.aliyun-inc.com/api";
         private static String project = "odps_public_dev";
         private static String table = "tunnel_sample_test";
         private static String partition = "pt=20150801,dt=hangzhou";
         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);
                         UploadSession uploadSession = tunnel.createUploadSession(project,
                                         table, partitionSpec);
                         System.out.println("Session Status is : "
                                         + uploadSession.getStatus().toString());
                         TableSchema schema = uploadSession.getSchema();
                         RecordWriter recordWriter = uploadSession.openRecordWriter(0);
                         Record record = uploadSession.newRecord();
                         for (int i = 0; i < schema.getColumns().size(); i++) {
                                 Column column = schema.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());
                                 }
                         }
                         for (int i = 0; i < 10; i++) {
                                 recordWriter.write(record);
                         }
                         recordWriter.close();
                         uploadSession.commit(new Long[]{0L});
                         System.out.println("upload success!");
                 } catch (TunnelException e) {
                         e.printStackTrace();
                 } catch (IOException e) {
                         e.printStackTrace();
                 }
         }
   }

   注解：這裏我們省略了accessId和accesskey的配置，實際運行時請換上您自己的accessId以及accessKey。

4. pom.xml文件配置如下：

   <?xml version="1.0" encoding="UTF-8"?>
   <project xmlns="https://maven.apache.org/POM/4.0.0"
   xmlns:xsi="https://www.w3.org/2001/XMLSchema-instance"
   xsi:schemaLocation="https://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.aliyun.odps.tunnel.example</groupId>
    <artifactId>UploadSample</artifactId>
    <version>1.0-SNAPSHOT</version>
    <dependencies>
      <dependency>
        <groupId>com.aliyun.odps</groupId>
        <artifactId>odps-sdk-core</artifactId>
        <version>0.20.7-public</version>
      </dependency>
    </dependencies>
    <repositories>
      <repository>
      <id>alibaba</id>
      <name>alibaba Repository</name>
      <url>https://mvnrepo.alibaba-inc.com/nexus/content/groups/public/</url>
      </repository>
    </repositories>
   </project>

5. 編譯與運行:編譯UploadSample工程：

   mvn package

   運行UploadSample程序，這裏我們利用eclipse導入maven project:右擊java工程並點擊< Import->Maven->Existing Maven Projects >設置如下：

Session Status is : NORMAL
upload success!

$ gem install fluent-plugin-aliyun-odps

$ gem install protobuf
$ gem install fluentd --no-ri --no-rdoc
$ git clone https://github.com/aliyun/aliyun-odps-fluentd-plugin.git
$ cp aliyun-odps-fluentd-plugin/lib/fluent/plugin/* {YOUR_FLUENTD_DIRECTORY}/lib/fluent/plugin/ -r

<source>
   type tail
   path /opt/log/in/in.log
   pos_file /opt/log/in/in.log.pos
   refresh_interval 5s
   tag in.log
   format /^(?<remote>[^ ]*) - - [(?<datetime>[^]]*)] "(?<method>S+)(?: +(?<path>[^"]*?)(?: +S*)?)?" (?<code>[^ ]*) (?<size>[^ ]*) "-" "(?<agent>[^"]*)"$/
   time_format %Y%b%d %H:%M:%S %z
</source>

<match in.**>
  type aliyun_odps
  aliyun_access_id ************
  aliyun_access_key *********
  aliyun_odps_endpoint https://service.odps.aliyun.com/api
  aliyun_odps_hub_endpoint https://dh.odps.aliyun.com
  buffer_chunk_limit 2m
  buffer_queue_limit 128
  flush_interval 5s
  project projectforlog
  <table in.log>
    table nginx_log
    fields remote,method,path,code,size,agent
    partition ctime=${datetime.strftime('%Y%m%d')}
    time_format %d/%b/%Y:%H:%M:%S %z
   </table>
</match>

$ gem install fluent-plugin-sql

<source>
  type sql
  host 127.0.0.1
  database test
  adapter mysql
  username xxxx
  password xxxx
  select_interval 10s
  select_limit 100
  state_file /path/sql_state
  <table>
    table test_table
    tag in.sql
    update_column id
  </table>
</source>

<match in.**>
  type aliyun_odps
  aliyun_access_id ************
  aliyun_access_key *********
  aliyun_odps_endpoint https://service.odps.aliyun.com/api
  aliyun_odps_hub_endpoint https://dh.odps.aliyun.com
  buffer_chunk_limit 2m
  buffer_queue_limit 128
  flush_interval 5s
  project your_projectforlog
  <table in.log>
    table mysql_data
    fields id,field1,field2,fields3
    </table>
</match>