云原生大数据计算服务 MaxCompute：查询加速（MCQA）

MaxCompute MCQA功能提供如下能力。

• 支持对中、小数据量查询作业进行加速优化，将执行时间为分钟级的查询作业缩减至秒级，同时完全兼容原MaxCompute的查询功能。

• 支持主流BI工具，开展即席查询（Ad Hoc）或商业智能（BI）分析。

• 支持使用独立的资源池，不占用离线计算资源，可以自动识别查询作业，缓解排队压力，优化使用体验。

• 支持将MCQA（MaxCompute Query Acceleration）查询作业的运行结果写入临时缓存中。当用户后续执行相同的查询作业时，MaxCompute会优先返回缓存中的结果，加快执行速度。

MCQA架构图如下所示。

MCQA功能的应用场景如下。

• 仅支持数据查询（SELECT开头）语句。如果您提交了MCQA不支持的语句，MaxCompute客户端、JDBC和SDK可通过配置回退到普通离线模式执行，其他工具暂不支持回退到普通离线模式执行。版本要求如下：

  • MaxCompute客户端（odpscmd）要求v0.40.8及以上版本，下载odpscmd。

  • ODPS-JDBC要求v3.3.0及以上版本，下载ODPS-JDBC。

  • ODPS-SDK版本v0.40.8-public及以上要求，下载ODPS-SDK。

• 默认最大查询100W行数据，可通过在SQL语句中增加Limit关键字突破此限制。

MCQA功能的详细使用限制如下。

出现如下情况时，MaxCompute会删除缓存结果：

• MaxCompute项目的资源使用率较高时，MaxCompute会提前删除缓存结果。

• 缓存结果所引用的表或视图变更后，缓存结果立即失效，MaxCompute会删除失效的缓存结果。

• 缓存结果已过期。

MCQA查询结果缓存的使用限制如下：

• 如果您希望从缓存的查询结果中检索数据，执行的查询作业以及查询的上下文配置必须与原始查询作业完全相同。当您运行重复的查询作业时，MaxCompute会重复使用缓存结果。

• 缓存的查询结果中所引用的表或视图变更后，缓存失效，再次执行相同的查询作业，不会检索到缓存数据。

• 临时数据集的缓存大小限制为10 GB。

您可以获取查询作业的Logview信息，并在Job Details页签，查看到查询作业运行结果已写入缓存中，如下图示所示。更多获取Logview信息，请参见使用Logview 2.0查看作业运行信息。

使用查询加速功能加速包年包月MaxCompute实例的项目步骤如下。

1. 登录MaxCompute控制台，在左上角选择地域。

2. 在左侧导航栏，选择工作区 > 配额（Quota）管理。

3. 在Quota管理页面，单击需要配置的一级Quota操作列的Quota配置。

4. 配置Quota模板。

   1. 在Quota配置页面的Quota模板页签，单击新增Quota。

   2. 在新增Quota对话框，单击+添加一项后，填写Quota名称和选择类型。

      Quota名称请自定义输入，类型选择交互式。

   3. 单击确定，完成新增MCQA资源组。

5. 配置Quota计划。

   1. 在Quota配置页面的Quota计划页签，可以单击添加计划来创建一个新的计划，也可以在已有计划后单击编辑做编辑操作。

   2. 在新建Quota计划或编辑Quota计划的对话框中，在MCQA资源组上输入预留CU量[minCU,maxCU]的值。

      输入预留CU量[minCU,maxCU]的值时注意事项如下。

      • 最小CU数（minCU）需要等于最大CU数（maxCU）。

      • 最小CU数要大于等于50CU。如果不需要交互式资源，设置为0。

      • 交互式类型的Quota不支持弹性预留CU。

   3. 单击确定，完成Quota计划录入。

   4. 在Quota计划页签，单击新增或编辑Quota计划操作列的立即生效

6. 配置时间计划。

   配置时间计划设置每日不同时间点启用不同的Quota计划，以此实现对Quota配置的分时逻辑。

MCQA功能支持的接入方式如下：

• MaxCompute客户端。

• DataWorks临时查询或数据开发。

• JDBC。

• SDK。

• MaxCompute Studio。

• PyODPS。

• SQLAlchemy。

在临时查询模块执行查询作业，如果返回结果包含如下信息表明MCQA功能已开启。临时查询详情请参见创建临时查询。

在手动业务流程模块执行查询作业，返回结果包含如下信息表明MCQA功能已开启。手动业务流程详情请参见创建手动任务。

使用JDBC连接MaxCompute，您可以通过执行如下操作开启MCQA功能。使用JDBC连接MaxCompute的操作详情请参见JDBC使用说明。

1. 下载支持MCQA功能的JDBC_JAR或可编译的源代码。

2. 通过Maven方式配置Pom依赖。

   <dependency>
     <groupId>com.aliyun.odps</groupId>
     <artifactId>odps-jdbc</artifactId>
     <version>3.3.0</version>
     <classifier>jar-with-dependencies</classifier>
   </dependency>

3. 基于源代码创建Java程序，适配实际信息，详情请参见MaxCompute JDBC，示例如下。

   // 阿里云账号AccessKey拥有所有API的访问权限，风险很高。强烈建议您创建并使用RAM用户进行API访问或日常运维，请登录RAM控制台创建RAM用户
   // 此处以把AccessKey 和 AccessKeySecret 保存在环境变量为例说明。您也可以根据业务需要，保存到配置文件里
   // 强烈建议不要把 AccessKey 和 AccessKeySecret 保存到代码里，会存在密钥泄漏风险
   private static String accessId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
   private static String accessKey = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
   //your_project_name为需要使用MCQA功能的项目名称。
   String conn = "jdbc:odps:http://service.<regionid>.maxcompute.aliyun.com/api?project=<YOUR_PROJECT_NAME>"&accessId&accessKey&charset=UTF-8&interactiveMode=true&alwaysFallback=false&autoSelectLimit=1000000000";
   Statement stmt = conn.createStatement();
   Connection conn = DriverManager.getConnection(conn, accessId, accessKey);
   Statement stmt = conn.createStatement();
   String tableName = "testOdpsDriverTable";
   stmt.execute("DROP TABLE IF EXISTS " + tableName);
   stmt.execute("CREATE TABLE " + tableName + " (key int, value string)");

   您可以选择在连接串中配置如下参数，完善处理逻辑。

   参数	说明
   enableOdpsLogger	用于打印日志。未配置SLF4J时，建议您配置此参数为True。
   fallbackForUnknownError	默认值为False，设置为True时，表示发生未知错误时回退到离线模式。
   fallbackForResourceNotEnough	默认值为False，设置为True时，表示发生资源不足问题时回退到离线模式。
   fallbackForUpgrading	默认值为False，设置为True时，表示升级期间回退到离线模式。
   fallbackForRunningTimeout	默认值为False，设置为True时，表示执行超时时回退到离线模式。
   fallbackForUnsupportedFeature	默认值为False，设置为True时，表示遇到MCQA不支持的场景时回退到离线模式。
   alwaysFallback	默认值为False，设置为True时，表示在以上几种场景下全部回退到离线模式，仅在JDBC 3.2.3及以上版本支持。

使用示例。

• 示例1：Tableau上使用MCQA：

  服务器增加interactiveMode=true属性，用于开启MCQA功能。建议您同步增加enableOdpsLogger=true属性，用于打印日志。配置操作详情请参见配置JDBC使用Tableau。

  完整的服务器配置示例如下。

  http://service.cn-beijing.maxcompute.aliyun.com/api?
  project=****_beijing&interactiveMode=true&enableOdpsLogger=true&autoSelectLimit=1000000000"

  如果只对项目空间中的部分表进行Tableau操作，您可以在服务器参数中增加table_list=table_name1, table_name2属性选择需要的表，表之间用半角逗号（,）分隔。如果表过多，会导致Tableau打开缓慢，强烈建议使用此方式只载入需要的表。示例如下，对于有大量分区的表不建议把所有分区的数据都设置成数据源，可以筛选需要的分区或通过自定义SQL获取需要的数据。

  http://service.cn-beijing.maxcompute.aliyun.com/api?project=****_beijing
  &interactiveMode=true&alwaysFallback=true&enableOdpsLogger=true&autoSelectLimit=1000000000"
  &table_list=orders,customers

• 示例2：SQLWorkBench使用MCQA。

  完成JDBC驱动配置后，在Profile配置界面修改已填写的JDBC URL，支持SQLWorkbench使用MCQA功能。Profile配置操作详情请参见配置JDBC使用SQL Workbench/J。

  需要配置的URL格式如下所示：

  jdbc:odps:<MaxCompute_endpoint>?
  project=<MaxCompute_project_name>&accessId=<AccessKey ID>&accessKey=<AccessKey Secret>
  &charset=UTF-8&interactiveMode=true&autoSelectLimit=1000000000"

  参数说明如下。

  参数	说明
  MaxCompute_endpoint	MaxCompute服务所在区域的Endpoint，详情请参见Endpoint。
  MaxCompute_project_name	MaxCompute项目空间名称。
  AccessKey ID	有访问指定项目空间权限的AccessKey ID。 您可以进入AccessKey管理页面获取AccessKey ID。
  AccessKey Secret	AccessKey ID对应的AccessKey Secret。 您可以进入AccessKey管理页面获取AccessKey Secret。
  charset=UTF-8	字符集编码格式。
  interactiveMode	MCQA功能开关，true表示开启MCQA功能。
  autoSelectLimit	数据量超过100万限制时，需要配置此参数。

Java SDK详情请参见Java SDK介绍。您需要通过Maven配置Pom依赖，配置示例如下。

<dependency>
  <groupId>com.aliyun.odps</groupId>
  <artifactId>odps-sdk-core</artifactId>
  <version>3.3.0</version>
</dependency>

创建Java程序，命令示例如下。

import com.aliyun.odps.Odps;
import com.aliyun.odps.OdpsException;
import com.aliyun.odps.OdpsType;
import com.aliyun.odps.account.Account;
import com.aliyun.odps.account.AliyunAccount;
import com.aliyun.odps.data.Record;
import com.aliyun.odps.data.ResultSet;
import com.aliyun.odps.sqa.*;

import java.io.IOException;
import java.util.*;

public class SQLExecutorExample {

    public static void SimpleExample() {
        // 设置账号和项目信息。
      	// 阿里云账号AccessKey拥有所有API的访问权限，风险很高。强烈建议您创建并使用RAM用户进行API访问或日常运维，请登录RAM控制台创建RAM用户
				// 此处以把AccessKey 和 AccessKeySecret 保存在环境变量为例说明。您也可以根据业务需要，保存到配置文件里
				// 强烈建议不要把 AccessKey 和 AccessKeySecret 保存到代码里，会存在密钥泄漏风险
        Account account = new AliyunAccount(System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
        Odps odps = new Odps(account);
        odps.setDefaultProject("<YOUR_PROJECT_NAME>");
        odps.setEndpoint("http://service.<regionid>.maxcompute.aliyun.com/api");

        // 准备构建SQLExecutor。
        SQLExecutorBuilder builder = SQLExecutorBuilder.builder();

        SQLExecutor sqlExecutor = null;
        try {
            // run in offline mode or run in interactive mode
            if (false) {
                // 创建一个默认执行离线SQL的Executor。
                sqlExecutor = builder.odps(odps).executeMode(ExecuteMode.OFFLINE).build();
            } else {
                // 创建一个默认执行查询加速SQL的Executor，并且在查询加速模式失败后，自动回退到离线查询。
                sqlExecutor = builder.odps(odps).executeMode(ExecuteMode.INTERACTIVE).fallbackPolicy(FallbackPolicy.alwaysFallbackPolicy()).build();
            }
            // 如果需要的话可以传入查询的特殊设置。
            Map<String, String> queryHint = new HashMap<>();
            queryHint.put("odps.sql.mapper.split.size", "128");
            // 提交一个查询作业，支持传入Hint。
            sqlExecutor.run("select count(1) from test_table;", queryHint);

            // 列举一些支持的常用获取信息的接口。
            // UUID
            System.out.println("ExecutorId:" + sqlExecutor.getId());
            // 当前查询作业的logview。
            System.out.println("Logview:" + sqlExecutor.getLogView());
            // 当前查询作业的Instance对象(Interactive模式多个查询作业可能为同一个Instance)。
            System.out.println("InstanceId:" + sqlExecutor.getInstance().getId());
            // 当前查询作业的阶段进度(Console的进度条)。
            System.out.println("QueryStageProgress:" + sqlExecutor.getProgress());
            // 当前查询作业的执行状态变化日志，例如回退信息。
            System.out.println("QueryExecutionLog:" + sqlExecutor.getExecutionLog());

            // 提供两种获取结果的接口。
            if(false) {
                // 直接获取全部查询作业结果，同步接口，可能会占用本线程直到查询成功或失败。
                // 一次性读取全部结果数据到内存中，当数据量较大时不建议使用，可能会有内存问题。
                List<Record> records = sqlExecutor.getResult();
                printRecords(records);
            } else {
                // 获取查询结果的迭代器ResultSet，同步接口，可能会占用本线程直到查询成功或失败。
                // 获取大量结果数据时推荐使用，分次读取查询结果。
                ResultSet resultSet = sqlExecutor.getResultSet();
                while (resultSet.hasNext()) {
                    printRecord(resultSet.next());
                }
            }

            // run another query
            sqlExecutor.run("select * from test_table;", new HashMap<>());
            if(false) {
                // 直接获取全部查询结果，同步接口，可能会占用本线程直到查询成功或失败。
                // 一次性读取全部结果数据到内存中，当数据量较大时不建议使用，可能会有内存问题。
                List<Record> records = sqlExecutor.getResult();
                printRecords(records);
            } else {
                // 获取查询结果的迭代器ResultSet，同步接口，可能会占用本线程直到查询成功或失败。
                // 获取大量结果数据时推荐使用，分次读取查询结果。
                ResultSet resultSet = sqlExecutor.getResultSet();
                while (resultSet.hasNext()) {
                    printRecord(resultSet.next());
                }
            }
        } catch (OdpsException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (sqlExecutor != null) {
                // 关闭Executor释放相关资源。
                sqlExecutor.close();
            }
        }
    }

    // SQLExecutor can be reused by pool mode
    public static void ExampleWithPool() {
        // 设置账号和项目信息。
      	// 阿里云账号AccessKey拥有所有API的访问权限，风险很高。强烈建议您创建并使用RAM用户进行API访问或日常运维，请登录RAM控制台创建RAM用户
				// 此处以把AccessKey 和 AccessKeySecret 保存在环境变量为例说明。您也可以根据业务需要，保存到配置文件里
				// 强烈建议不要把 AccessKey 和 AccessKeySecret 保存到代码里，会存在密钥泄漏风险
        Account account = new AliyunAccount(System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
        Odps odps = new Odps(account);
        odps.setDefaultProject("your_project_name");
        odps.setEndpoint("http://service.<regionid>.maxcompute.aliyun.com/api");

        // 通过连接池方式执行查询。
        SQLExecutorPool sqlExecutorPool = null;
        SQLExecutor sqlExecutor = null;
        try {
            // 准备连接池，设置连接池大小和默认执行模式。
            SQLExecutorPoolBuilder builder = SQLExecutorPoolBuilder.builder();
            builder.odps(odps)
                    .initPoolSize(1) // init pool executor number
                    .maxPoolSize(5)  // max executors in pool
                    .executeMode(ExecuteMode.INTERACTIVE); // run in interactive mode

            sqlExecutorPool = builder.build();
            // 从连接池中获取一个Executor，如果不够将会在Max限制内新增Executor。
            sqlExecutor = sqlExecutorPool.getExecutor();

            // Executor具体用法和上一示例一致。
            sqlExecutor.run("select count(1) from test_table;", new HashMap<>());
            System.out.println("InstanceId:" + sqlExecutor.getId());
            System.out.println("Logview:" + sqlExecutor.getLogView());

            List<Record> records = sqlExecutor.getResult();
            printRecords(records);
        } catch (OdpsException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            sqlExecutor.close();
        }
        sqlExecutorPool.close();
    }

    private static void printRecord(Record record) {
        for (int k = 0; k < record.getColumnCount(); k++) {

            if (k != 0) {
                System.out.print("\t");
            }

            if (record.getColumns()[k].getType().equals(OdpsType.STRING)) {
                System.out.print(record.getString(k));
            } else if (record.getColumns()[k].getType().equals(OdpsType.BIGINT)) {
                System.out.print(record.getBigint(k));
            } else {
                System.out.print(record.get(k));
            }
        }
    }

    private static void printRecords(List<Record> records) {
        for (Record record : records) {
            printRecord(record);
            System.out.println();
        }
    }

    public static void main(String args[]) {
        SimpleExample();
        ExampleWithPool();
    }
}

您在MaxCompute Studio的SQL编辑器中，选择带查询加速功能的SQL执行模式查询加速或加速失败重跑后，执行查询语句即可启用查询加速功能。

• 查询加速：使用查询加速功能执行SQL查询语句。

• 加速失败重跑：使用查询加速功能执行SQL查询语句失败后，回退到离线模式（即默认模式）执行SQL查询语句。

PyODPS集成了SQLAlchemy，可以使用SQLAlchemy查询MaxCompute数据。您需要在连接串中指定如下参数实现查询加速：

• interactive_mode=true：必填。查询加速功能总开关。

• reuse_odps=true：可选。打开强制复用连接，对于部分第三方工具（例如Apache Superset），打开此选项可提高性能。

您可以在连接串中配置fallback_policy=<policy1>,<policy2>,...参数，完善处理逻辑。与JDBC的配置项类似，控制加速失败的回退行为。

• generic：默认为False，设置为True时，表示发生未知错误时回退到离线模式。

• noresource：默认为False，设置为True时，表示发生资源不足问题时回退到离线模式。

• upgrading：默认为False，设置为True时，表示升级期间回退到离线模式。

• timeout：默认为False，设置为True时，表示执行超时时回退到离线模式。

• unsupported：默认为False，设置为True时，表示遇到MCQA不支持的场景时回退到离线模式。

• default：等同于同时指定unsupported、upgrading、noresource和timeout。如果连接串中未指定fallback_policy，则此项为默认值。

• all：默认为False，设置为True时，表示在以上几种场景下全部回退到离线模式。

例如，打开查询加速，打开强制复用链接，在查询加速功能尚未支持、升级中和资源不足时回退到离线模式的连接串如下。

odps://<access_id>:<ACCESS_KEY>@<project>/?endpoint=<endpoint>&interactive_mode=true&reuse_odps=true&fallback_policy=unsupported,upgrading,noresource