MaxCompute查询加速MCQA是什么- 云原生大数据计算服务 ...

本文为您介绍MaxCompute查询加速MCQA（MaxCompute Query Acceleration）功能，并帮助您了解该功能的系统架构、关键特性、应用场景和使用限制。

功能介绍

MaxCompute MCQA功能提供如下能力。

支持对中、小数据量查询作业进行加速优化，将执行时间为分钟级的查询作业缩减至秒级，同时完全兼容原MaxCompute的查询功能。
支持主流BI工具，开展即席查询（Ad Hoc）或商业智能（BI）分析。
支持使用独立的资源池，不占用离线计算资源，可以自动识别查询作业，缓解排队压力，优化使用体验。
支持将MCQA（MaxCompute Query Acceleration）查询作业的运行结果写入临时缓存中。当用户后续执行相同的查询作业时，MaxCompute会优先返回缓存中的结果，加快执行速度。

场景	说明	场景特点
即席查询（Ad Hoc）	您可以通过MCQA优化中小规模数据集（百GB规模内）的查询性能，直接对MaxCompute表开展低时延的查询操作，以便快速完成数据开发及数据分析。	您可以根据实际需求，灵活选择查询条件，快速获取查询结果并调整查询逻辑。对查询时延的要求在几十秒内。使用者通常为掌握SQL技能的数据开发或数据分析师，希望使用熟悉的客户端工具开展查询分析。
商业智能（BI）	利用MaxCompute搭建企业级数据仓库时，ETL会将数据加工处理为面向业务可消费的聚合数据。借助MCQA的低延时、弹性并发、数据缓存等特性，结合MaxCompute表分区、分桶等优化设计，可以低成本满足多并发、快速响应的报告生成、统计分析及固定报表分析需求。	查询的数据对象通常为聚合后的结果数据，适用于数据量较小、多维查询、固定查询、高频查询场景。查询延时要求高，秒级返回（例如大部分查询不超过5秒，不同查询作业由于数据规模和查询复杂度不同，查询时间有较大差异）。
海量数据明细查询分析	MCQA可以自动识别查询作业特征，既能快速响应，处理小规模作业，同时还可以自动匹配大规模作业资源需求，满足分析人员分析不同规模和复杂度的查询作业的需求。	需要探索的历史数据量大、真正需要的有效数据量不大、查询延时要求适中。使用者通常为业务分析人员，往往需要从明细数据中探寻业务规律，发现业务机会，验证业务假设。

限制项	说明
功能	MCQA功能支持标准版、按量计费的MaxCompute服务。 MCQA功能支持包年包月计费模式。 MCQA功能不支持开发者版的MaxCompute服务，需要升级为标准版。
查询	单次执行的作业支持并发的Worker数上限为2000个。您通过客户端提交的MCQA作业默认30s超时，通过DataWorks临时查询提交的MCQA作业默认20s超时，超时后MCQA作业将默认回退为普通查询作业。仅存储格式为ALIORC的表的数据才可能被缓存至内存中进行加速。
查询并发	包年包月模式。免费版（未划分MCQA交互式资源组）。每天每个项目并发限制5个，每天每个项目最多运行500个作业，超过部分默认设置自动回退为普通作业，如果您设置了不回退，系统会报如下错误： `ODPS-1800001: Session exception - Failed to submit sub-query in session because:Prepaid project run outoffree query quota.` MCQA交互式资源组。每个项目的MCQA作业最大并发数为120，超过后提交作业会回退到普通作业模式运行。配置交互式资源类型的配额组，用以MCQA作业的预留CU最小配额需和预留CU最大配额值需一致，否则不生效。交互式资源类型需要满足以下要求，否则不能提交。 `预留CU量[minCU]` = `预留CU量[maxCU]` 。预留CU量需大于等于 `50` CU。配置了交互式资源配额组后，所有Project仅支持可识别为加速的作业提交到交互式资源配额组，且查询加速功能不再享有免费体验。交互式资源类型的配额组不支持作为Project的默认Quota使用，使用查询加速功能时交互式配额组无需绑定Project即对所有Project生效。按量付费模式。单个MaxCompute项目的MCQA作业最大并发数为120，超过部分会回退到普通作业模式运行。

SQLExecutorBuilder builder = SQLExecutorBuilder.builder();
builder.quotaName("<OfflineQuotaName>");

enable_interactive_mode=true --打开MCQA
interactive_auto_rerun=true  --代表MCQA失败后自动回退到普通作业执行

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

// 阿里云账号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及以上版本支持。

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

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万限制时，需要配置此参数。

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

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();
}

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

```
sError:com.aliyun.odps.OdpsException: ODPS-1800001: Session exception - Failed to submit sub-query in session because:Prepaid project run out of free query quota.
```

from odps import ODPS, errors
max_retry_times = 3
def run_sql(odps, stmt):
    retry = 0
    while retry < max_retry_times:
            inst = odps.run_sql_interactive(stmt)
            print(inst.get_logview_address())
            inst.wait_for_success(interval=0.1)
            records = []
            for each_record in inst.open_reader(tunnel=True):
                records.append(each_record)
            return records
        except errors.ODPSError as e:
            retry = retry + 1
            print("Error: " + str(e) + " retry: " + str(retry) + "/" + str(max_retry_times))
            if retry >= max_retry_times:
                raise e
odps = ODPS(...)
run_sql(odps, 'SELECT 1')

```
String logview = sqlExecutor.getLogView();
```

功能介绍

产品架构

应用场景

使用限制

缓存机制

使用限制

缓存计费规则

缓存验证

包年包月规格开通MCQA

操作步骤

调度策略

回退策略

MCQA渠道接入说明

基于MaxCompute客户端启用MCQA功能

基于DataWorks临时查询或数据开发启用MCQA功能

基于JDBC启用MCQA功能

基于Java SDK启用MCQA功能

基于MaxCompute Studio启用MCQA功能

基于PyODPS使用MCQA功能

基于PyODPS，使用SQLAlchemy或其他支持SQLAlchemy接口的第三方工具实现查询加速

常见问题