使用说明

• 使用JDBC连接MaxCompute，详情请参见JDBC使用说明。如果您需要在该场景下开启MCQA功能，需要修改相关配置，详情请参见配置JDBC启用MCQA功能。
• 使用JDBC连接Tableau，对MaxCompute中的数据进行可视化分析，详情请参见配置JDBC使用Tableau。如果您需要在该场景下开启MCQA功能，需要修改相关配置，详情请参见基于JDBC配置Tableau启用MCQA功能。
• 使用JDBC连接SQL Workbench/J，使用SQL Workbench/J对MaxCompute中的数据执行SQL语句，详情请参见配置JDBC使用SQL Workbench/J。如果您需要在该场景下开启MCQA功能，需要修改相关配置，详情请参见基于JDBC配置SQLWorkBench启用MCQA功能。

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&enableOdpsLogger=true&autoSelectLimit=1000000000"&table_list=orders,customers

• maxcompute_endpoint：MaxCompute服务所在区域的Endpoint。详情请参见Endpoint。
• maxcompute_project_name：MaxCompute项目空间名称。
• AccessKey ID：有访问指定项目空间权限的AccessKey ID。
• AccessKey Secret：AccessKey ID对应的AccessKey Secret。
• charset=UTF-8：字符集编码格式。
• interactiveMode：MCQA功能开关，true表示开启MCQA功能。
• autoSelectLimit：数据量超过100万限制时，需要配置此参数。

<dependency>
  <groupId>com.aliyun.odps</groupId>
  <artifactId>odps-sdk-core</artifactId>
  <version>0.35.5-public</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();
    }
}

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