使用 Instance Tunnel 获取 Maxcompute Instance 的执行结果

  1. 云栖社区>
  2. 博客>
  3. 正文

使用 Instance Tunnel 获取 Maxcompute Instance 的执行结果

贞宏 2017-07-24 17:50:07 浏览2530

本篇将介绍如何使用 Instance Tunnel 来获取 Maxcompute Instance 执行结果。

源起

每天我们都会在 Maxcompute 平台上提交 select query,用于查询特定的数据。然而,熟悉平台的同学都知道,从平台获取 sql 查询结果是一个 Restful 请求,可能碰到以下两个问题:

  • 1 获取数据超时。 如果数据分布在多个存储小文件上,平台需要花费大量时间来收集和归并这些数据。然而在这个漫长的归并过程中,获取数据的 Restful 请求可能已经超时了。此时 Maxcompute Console 会有如下警告:Warning: ODPS request failed, requestID:xxxx, retryCount:1, will retry in xxx seconds.
  • 2 获取数据量受限。 由于一次 Restful 请求的返回数据有限,且一次性获取全量数据到本地时可能将内存撑爆等问题,Maxcompute SQL 的查询结果条数是受限的,具体的数值为 project 上的配置项 READ_TABLE_MAX_ROW (默认为 10000)。这就会出现明明我们的查询结果 2W 条,最后却在 Maxcompute Console 或者 Logview 上只看到 1W条 的诡异情况了。

针对上述问题,提出下列解法,其中解法 3 可解决上述两种问题,解法 1、2 仅适用于问题 1:

  • 1 合并存储小文件。 如果查询语句的数据源只有一张表,那么可以在 Maxcompute Console 执行alter table <source_table_name> merge smallfiles ; 命令将小文件合并之后,再重新执行 select 查询。
  • 2 创建临时表,合并临时表的存储小文件。 如果是一个多表计算的查询结果,可以通过 create table <tmp_table_name> as select ....命令创建临时表,再参照解法 a 合并临时表的存储小文件。
  • 3 创建临时表,使用 Tunnel 下载。 Tunnel 是 Maxcompute 批量数据通道。因此我们可以通过 create table <tmp_table_name> as select ....命令创建临时表,然后通过 Tunnel 命令或者 Tunnel SDK 来将临时表数据下载下来。

Instance Tunnel 概述

上面的方法虽然能解决问题,但总是有那么点"绕",有那么点"费"。

有没有更加直接的方法呢?
如果是之前,只能怂了,但是现在我们要大声的说有!

在最近的 Maxcompute 版本( >=S27 ) 中,我们开发了Instanc Tunnel功能。
Instance Tunnel 提供使用 Tunnel 来下载 SQL 查询结果的功能,不仅能摆脱上述两类问题,可直接获取查询结果;还丰富了 Maxcompute Tunnel 下载通道,不再局限于表数据。换句话说,以前我们可以用 Tunnel 来下载 Maxcompute 表数据,如今,我们也可以用 Tunnel 来下载 Maxcompute Instance 的数据。

Instance Tunnel 使用

Instance Tunnel 的使用方式与 Table Tunnel 基本一致,下面分别介绍使用客户端和 SDK 来下载 Instance 执行结果的方法。

1. 用 Maxcompute Console 来获取 Instance 数据

1.1 使用 Tunnel download 命令将特定 Instance 的执行结果下载到本地文件

命令:
tunnel download instance://<[project_name/]instance_id> <path>
参数:

project_name:  instance 所在的项目名称;
instance_id:  待下载数据的 instance id

举例:


// 执行一条 select 查询:
odps@ odps_test_project>select * from wc_in;
ID = 20170724071705393ge3csfb8
... ...

// 使用 Instance Tunnel Download 命令下载执行结果到本地文件
odps@ odps_test_project>tunnel download instance://20170724071705393ge3csfb8 result;
2017-07-24 15:18:47  -  new session: 2017072415184785b6516400090ca8    total lines: 8
2017-07-24 15:18:47  -  file [0]: [0, 8), result
downloading 8 records into 1 file
2017-07-24 15:18:47  -  file [0] start
2017-07-24 15:18:48  -  file [0] OK. total: 44 bytes
download OK

// 查看结果
   cat result
slkdfj
hellp
apple
tea
peach
apple
tea
teaa

1.2 通过配置参数使 SQL 查询默认采用 Instance Tunnel 方式输出执行结果

在 Maxcompute Console 中打开 use_instance_tunnel 选项之后,执行的 select query 就会默认使用 Instance tunnel 来下载结果了,再也不会出现文章开头所描述的两种问题了。打开该配置有两种方法:

  • 1) 如果已经下载最新的 Console,odps_config.ini 里面已经默认打开此选项,并默认将 instance_tunnel_max_record 设置成了10000 。 如下所示:
# download sql results by instance tunnel
use_instance_tunnel=true
# the max records when download sql results by instance tunnel
instance_tunnel_max_record=10000

其中 instance_tunnel_max_record 表示使用 Instance tunnel 下载 sql 查询结果的条数。若不设置,下载条数不受限。

  • 2) 使用 set console.sql.result.instancetunnel=true; 开启此功能。
// 打开 Instance tunnel 选项
odps@ odps_test_tunnel_project>set console.sql.result.instancetunnel=true;
OK

// 运行 select query
odps@ odps_test_tunnel_project>select * from wc_in;
ID = 20170724081946458g14csfb8
Log view:
http://logview/xxxxx.....
+------------+
| key        |
+------------+
| slkdfj     |
| hellp      |
| apple      |
| tea        |
| peach      |
| apple      |
| tea        |
| teaa       |
+------------+
A total of 8 records fetched by instance tunnel.

可以看到,如果使用 Instance tunnel 的方式来输出 select 查询结果,会在最后打印一条提示。比如上面例子中的提示告诉我们这个 instance 的执行结果一共有 8 条数据。同样也可以 set console.sql.result.instancetunnel=false;来关闭此功能。

2. 用 MaxCompute Instance Tunnel SDK 来获取 Instance 执行结果

MaxCompute Java SDK 和 Python SDK 都对 Instance tunnel 进行了支持,下面介绍用法。

2.1 使用 Java SDK 获取 Instance 执行结果

对于 Java SDK,从 0.27.2-public (JavaDoc )版本开始,我们提供两种方式来获取数据。

  • 使用 SQLTask.getResultSet() 静态方法获取:
   Odps odps = OdpsUtils.newDefaultOdps(); // 初始化 Odps 对象
    Instance i = SQLTask.run(odps, "select * from wc_in;");
    i.waitForSuccess();
    
    // 根据 instance 对象,获取结果迭代器
    ResultSet rs = SQLTask.getResultSet(i);
    for (Record r : rs) {
      // 输出结果条数
      System.out.println(rs.getRecordCount());

      for (int col = 0; col < rs.getTableSchema().getColumns().size(); ++col) {
        // wc_in 表字段均为 STRING, 这里就直接打印输出
        System.out.println(r.get(col));
      }
    }
  • 创建 InstanceTunnel.DownloadSession 来获取:
   Odps odps = OdpsUtils.newDefaultOdps(); // 初始化 Odps 对象
    Instance i = SQLTask.run(odps, "select * from wc_in;");
    i.waitForSuccess();
    
    // 创建 InstanceTunnel
    InstanceTunnel tunnel = new InstanceTunnel(odps);
    // 根据 instance id,创建 DownloadSession
    InstanceTunnel.DownloadSession session = tunnel.createDownloadSession(odps.getDefaultProject(), i.getId());

    long count = session.getRecordCount();
     // 输出结果条数
    System.out.println(count);

    // 获取数据的写法与 TableTunnel 一样
    TunnelRecordReader reader = session.openRecordReader(0, count);
    Record record;
    while ((record = reader.read()) != null) {
      for (int col = 0; col < session.getSchema().getColumns().size(); ++col) {
        // wc_in 表字段均为 STRING, 这里就直接打印输出
        System.out.println(record.get(col));
      }
    }
    reader.close();  

2.2 使用 PyODPS 获取 Instance 执行结果

对于 PyODPS 来说,我们可以在 instance 上通过 open_reader 来获取数据,而从 0.7.7.1 的版本开始,我们可以通过 open_reader 使用 instance tunnel 来获取全量数据。

instance = o.execute_sql('select * from movielens_ratings limit 20000')
with instance.open_reader() as reader:
    print(reader.count)
    # for record in reader 就是遍历这2万条数据,这里通过切片只取10条
    for record in reader[:10]:  
        print(record)

Instance Tunnel 约束

虽然 Instance Tunnel 为我们提供了非常方便的获取 Instance 执行结果的方式,但是为了保护用户数据安全,此功能也受到了诸多的权限约束:

  • 若使用 Instance Tunnel 下载数据时,数据条数不超过 1W,则只要对该 Instance 有 Read 权限即可使用。 此行为与使用 Restful API 获取查询数据的行为一致;
  • 若使用 Instance Tunnel 下载数据时,数据条数超过 1W,则需要对 instance sql 查询语句中涉及到的所有源表进行权限检查,用户需要具有这些表的 Read 权限才可。

相关资料

Instance Tunnel 刚刚上线,欢迎讨论和吐槽 ^_^

  • Java SDK 和客户端使用,可联系作者
  • PyOdps 钉钉群:11701793