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

2022-04-11 04:14:50

本篇将介绍如何使用 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
上一篇：Linux 下查看服务器CPU核数


下一篇：Fedora把命令行下的执行结果保存到文件