查询加速（MCQA）

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

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

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

3. 在Quota管理页面，单击目标Quota对应操作列的Quota配置。

4. Quota基础配置

   1. 在Quota配置页面，选择基础配置页签，单击编辑基础配置。

   2. 单击新增二级Quota或为现有的二级Quota配置基础参数。

      Quota名称自定义输入，类型选择交互式。详细参数参考配置Quota。

5. Quota伸缩配置

   1. 在Quota配置页面，选择伸缩配置页签。

   2. 在伸缩配置页签，单击新增配置方案或单击已有的配置方案对应的操作列的编辑，更新配置方案。

   3. 在新增配置方案或编辑配置方案对话框的MCQA资源组上配置预留CU量[minCU,maxCU]。

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

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

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

   4. 单击确定，完成Quota配置录入。

   5. 在伸缩配置页签，单击目标配置方案对应的操作列的立即生效，可使该配置方案立即生效。也可在分时管理配置时使用已有方案。

6. 配置时间计划。

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

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

• 方式一：基于MaxCompute客户端启用MCQA功能

• 方式二：基于DataWorks临时查询或数据开发启用MCQA功能

• 方式三：基于JDBC启用MCQA功能

• 方式四：基于Java SDK启用MCQA功能

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

• 方式六：基于PyODPS使用MCQA功能

方式三：基于JDBC启用MCQA功能

方式四：基于Java SDK启用MCQA功能

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