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時，表示在以上幾種情境下全部回退到離線模式，僅在JDBC 3.2.3及以上版本才支援。

例如，開啟查詢加速，開啟強制複用連結，在查詢加速功能尚未支援、升級中和資源不足時回退到離線模式的串連串如下。

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