MaxCompute:Query Acceleration (MCQA)

Follow these steps to use the query acceleration feature to accelerate projects in a subscription MaxCompute instance.

1. Log on to the MaxCompute console and select a region in the top-left corner.

2. In the navigation pane on the left, choose Manage Configurations > Quotas.

3. On the Quotas page, find the target quota and click Quota Configuration in the Actions column.

4. Configure basic quota settings.

   1. On the Quota Configuration page, select the Basic Configurations tab and click Edit Basic Configurations.

   2. Click Add Level-2 Quota or configure basic parameters for an existing level-2 quota.

      Enter a custom Quota Name and set Type to Interactive. For more information about the parameters, see Configure a quota.

5. Configure quota scaling.

   1. On the Quota Configuration page, select the Scaling Configuration tab.

   2. On the Scaling Configuration tab, click Add Configuration Plan or find an existing configuration plan and click Edit in the Actions column to update the plan.

   3. In the Add Configuration Plan or Edit Configuration Plan dialog box, configure Reserved CUs [minCU,maxCU] for the MCQA resource group.

      • The minimum number of CUs (minCU) must be equal to the maximum number of CUs (maxCU).

      • The minimum number of CUs must be greater than or equal to 50. If you do not need interactive resources, set this parameter to 0.

      • Elastic reserved CUs are not supported for interactive quotas.

   4. Click OK to save the quota configuration.

   5. On the Scaling Configuration tab, find the target configuration plan and click Apply Immediately in the Actions column to immediately apply the plan. You can also use an existing plan when you configure Scheduled Scaling Management.

6. Configure a time-based plan.

   You can configure a time-based plan to enable different quota plans at different times of the day. This implements time-based logic for quota configuration.

MCQA supports the following connection methods:

• Method 1: Enable MCQA using the MaxCompute client

• Method 2: Enable MCQA using the ad hoc query or data development feature in DataWorks

• Method 3: Enable MCQA using JDBC

• Method 4: Enable MCQA using the Java SDK

• Method 5: Use PyODPS with SQLAlchemy or other SQLAlchemy-compatible third-party tools to accelerate queries

Method 3: Enable MCQA using JDBC

Method 4: Enable MCQA using the Java SDK

For more information about the Java SDK, see Java SDK overview. You must configure the Pom dependency in Maven. The following code provides an example configuration.

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

The following is an example command to create a Java program.

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() {
    // Set the account and project information.
    // An Alibaba Cloud account AccessKey has full permissions on all APIs and poses a high security risk. We recommend that you create and use a RAM user to make API calls or perform O&M. To create a RAM user, log on to the RAM console.
    // This example shows how to store the AccessKey ID and AccessKey secret in environment variables. You can also store them in a configuration file as needed.
    // We strongly recommend that you do not hard-code the AccessKey ID and AccessKey secret in your code. This can lead to security risks.
    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");

    // Prepare to build the SQLExecutor.
    SQLExecutorBuilder builder = SQLExecutorBuilder.builder();

    SQLExecutor sqlExecutor = null;
    try {
      // run in offline mode or run in interactive mode
      if (false) {
        // Create an executor that runs offline SQL statements by default.
        sqlExecutor = builder.odps(odps).executeMode(ExecuteMode.OFFLINE).build();
      } else {
        // Create an executor that runs accelerated SQL queries by default and automatically falls back to offline query mode if the acceleration fails.
        sqlExecutor = builder.odps(odps).executeMode(ExecuteMode.INTERACTIVE).fallbackPolicy(FallbackPolicy.alwaysFallbackPolicy()).build();
      }
      // You can pass special settings for the query if needed.
      Map<String, String> queryHint = new HashMap<>();
      queryHint.put("odps.sql.mapper.split.size", "128");
      // Submit a query job. Hints are supported.
      sqlExecutor.run("select count(1) from test_table;", queryHint);

      // The following code provides some common methods for obtaining information.
      // UUID
      System.out.println("ExecutorId:" + sqlExecutor.getId());
      // The Logview of the current query job.
      System.out.println("Logview:" + sqlExecutor.getLogView());
      // The Instance object of the current query job. In interactive mode, multiple query jobs may share the same instance.
      System.out.println("InstanceId:" + sqlExecutor.getInstance().getId());
      // The stage progress of the current query job (progress bar in the console).
      System.out.println("QueryStageProgress:" + sqlExecutor.getProgress());
      // The execution status change log of the current query job, such as fallback information.
      System.out.println("QueryExecutionLog:" + sqlExecutor.getExecutionLog());

      // Two methods are provided to obtain results.
      if(false) {
        // Directly obtain all query results. This is a synchronous method and may occupy the current thread until the query succeeds or fails.
        // This method reads all result data into the memory at once. We do not recommend using this method for large datasets due to potential memory issues.
        List<Record> records = sqlExecutor.getResult();
        printRecords(records);
      } else {
        // Obtain an iterator (ResultSet) for the query results. This is a synchronous method and may occupy the current thread until the query succeeds or fails.
        // We recommend using this method to retrieve large datasets because it reads query results in batches.
        ResultSet resultSet = sqlExecutor.getResultSet();
        while (resultSet.hasNext()) {
          printRecord(resultSet.next());
        }
      }

      // run another query
      sqlExecutor.run("select * from test_table;", new HashMap<>());
      if(false) {
        // Directly obtain all query results. This is a synchronous method and may occupy the current thread until the query succeeds or fails.
        // This method reads all result data into the memory at once. We do not recommend using this method for large datasets due to potential memory issues.
        List<Record> records = sqlExecutor.getResult();
                printRecords(records);
            } else {
                // Obtain an iterator (ResultSet) for the query results. This is a synchronous method and may occupy the current thread until the query succeeds or fails.
                // We recommend using this method to retrieve large datasets because it reads query results in batches.
                ResultSet resultSet = sqlExecutor.getResultSet();
                while (resultSet.hasNext()) {
                    printRecord(resultSet.next());
                }
            }
        } catch (OdpsException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (sqlExecutor != null) {
                // Close the executor to release resources.
                sqlExecutor.close();
            }
        }
    }

    // SQLExecutor can be reused by pool mode
    public static void ExampleWithPool() {
        // Set the account and project information.
      	// An Alibaba Cloud account AccessKey has full permissions on all APIs and poses a high security risk. We recommend that you create and use a RAM user to make API calls or perform O&M. To create a RAM user, log on to the RAM console.
		// This example shows how to store the AccessKey ID and AccessKey secret in environment variables. You can also store them in a configuration file as needed.
		// We strongly recommend that you do not hard-code the AccessKey ID and AccessKey secret in your code. This can lead to security risks.
        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");

        // Execute a query using a connection pool.
        SQLExecutorPool sqlExecutorPool = null;
        SQLExecutor sqlExecutor = null;
        try {
            // Prepare the connection pool, set the pool size and default execution mode.
            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();
            // Get an executor from the connection pool. If the pool is empty, a new executor is created within the max limit.
            sqlExecutor = sqlExecutorPool.getExecutor();

            // The usage of the executor is the same as in the previous example.
            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();
    }
}