查询加速（MCQA）

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

• MaxCompute客户端（odpscmd）要求v0.40.8及以上版本，下载odpscmd。
• ODPS-JDBC要求v3.3.0及以上版本，下载ODPS-JDBC。
• ODPS-SDK版本v0.40.8-public及以上要求，下载ODPS-SDK。

• MaxCompute项目的资源使用率较高时，MaxCompute会提前删除缓存结果。
• 缓存结果所引用的表或视图变更后，缓存结果立即失效，MaxCompure会删除失效的缓存结果。
• 缓存结果已过期。

• 如果您希望从缓存的查询结果中检索数据，执行的查询作业以及查询的上下文配置必须与原始查询作业完全相同。当您运行重复的查询作业时，MaxCompute会重复使用缓存结果。
• 缓存的查询结果中所引用的表或视图变更后，缓存失效，再次执行相同的查询作业，不会检索到缓存数据。
• 临时数据集的缓存大小限制为10 GB。

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配置的分时逻辑，配置详情请参见配置Quota。

• MaxCompute客户端。
• DataWorks临时查询或数据开发。
• JDBC。
• SDK。
• MaxCompute Studio。
• PyODPS。
• SQLAlchemy。

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，示例如下。

   String accessId = "your_access_id";
   String accessKey = "your_access_key";
   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();
   --your_access_id、your_access_key为您的云账号信息，your_project_name为需要使用MCQA功能的项目名称。
   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万限制时，需要配置此参数。

<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() {
        // 设置账号和项目信息。
        Account account = new AliyunAccount("<your_access_id>", "<your_access_key>");
        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() {
        // 设置账号和项目信息。
        Account account = new AliyunAccount("your_access_id", "your_access_key");
        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();
    }
}

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

odps = ODPS(...)
instance = odps.run_sql_interactive('select count(*) from test_table')
# logview
print(instance.get_logview_address())
# 可选：如果使用Tunnel下载结果，则不需要客户端等待。
# 等待运行完毕，也可以使用wait_for_completion，区别在于wait_for_success将在任务失败时抛出异常。
instance.wait_for_success()
# 下载结果
# 显示ODPS返回的警告
print(instance.get_warnings())
# 用于屏显，若任务失败则返回错误信息
print(instance.get_printable_result())
# 从frontend下载结果
for each_record in instance.open_reader():
    pass # each_record 是 odps.Record 类型
# 使用tunnel下载结果
for each_record in instance.open_reader(tunnel=True):
    pass

• service_name：可选。用于指定查询加速会话的名称。
• service_startup_timeout：可选。用于指定等待Session Attach的超时时间。
• force_reattach：可选。run_sql_interactive()默认会自动复用Session。用于强制重新Attach Session。

• interactive_mode=true：必填。查询加速功能总开关。
• reuse_odps=true：可选。打开强制复用连接，对于部分第三方工具（例如Apache SuperSet），打开此选项可提高性能。

• 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时，表示在以上几种场景下全部回退到离线模式，仅在JDBC 3.2.3及以上版本才支持。

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