当使用场景中不关心整个结果集的顺序时，可以使用并发导出数据功能以更快的速度将命中的数据全部返回。
背景
多元索引中提供了Search接口，Search接口支持全功能集，包括所有的查询功能，以及排序、统计聚合等分析能力，其结果会按照指定的顺序返回。
但是在有些场景中，例如对接计算系统Spark、Presto等或者一些圈选场景，只需要使用完整的查询能力，能将命中的数据以更快的速度全部返回，不关心整个结果集的顺序。为了支持此类需求，多元索引中提供了ParallelScan接口。
说明 5.6.0及其以上版本的Java SDK开始支持ParallelScan功能。
ParallelScan接口相对于Search接口，保留了所有的查询功能，但是舍弃了排序、统计聚合等分析功能，带来了5倍以上的性能提升，因此可以实现1分钟内上亿级别数据行的导出能力，导出能力可以水平扩展，不存在上限。
在接口的实现中，单次请求的limit限制更宽松。目前Search接口的limit最大允许100行，但是ParallelScan接口的limit最大允许2000行。同时支持多个线程一起并发请求，因此导出速度会极快。
场景
- 如果请求关心排序、统计聚合，或者是终端客户的直接查询请求，需要用Search接口。
- 如果请求不关心排序，只关心把所有符合条件的数据尽快返回，或者是计算系统（Spark、Presto等）拉取数据，可以考虑ParallelScan接口。
特点
ParallelScan接口相对于Search接口的典型特征如下：
- 结果稳定性
ParallelScan任务是有状态的。在一个Session请求中，获取到的结果集是确定的，由发起第一次请求时的数据状态决定。如果发起第一次请求后插入了新的数据或者修改了原有的数据不会对结果集造成影响。
- 新增会话（Session）概念
在ParallelScan系列接口中新增了Session概念。使用sessionId能够保证获取到的结果集是稳定的，具体流程如下：
- 通过ComputeSplits接口获取最大并发数和当前sessionId。
- 通过发起多个并发ParallelScan请求读取数据，请求中需要指定当前的sessionId和当前并发ID。
在某些不易获取sessionId的场景中，ParallelScan也支持不携带sessionId发起请求，但是不使用sessionId可能会有极小的概率导致获取到的结果集中有重复数据。
如果在ParallelScan过程中发生网络异常、线程异常、动态Schema修改、索引切换等情况，导致ParallelScan不能继续扫描数据，服务端会返回“OTSSessionExpired”错误码，此时需要重新开始一个新的ParallelScan任务，从最开始重新拉取数据。
- 最大并发数
ParallelScan支持的单请求的最大并发数由ComputeSplits的返回值确定。数据越多，支持的并发数越大。
单请求指同一个查询语句，例如查询city=“杭州”的结果，如果使用Search接口，那么Search请求的返回值中会包括所有city=“杭州”的结果；如果使用ParallelScan接口且并发数是2，那么每个ParallelScan请求返回50%的结果，然后将两个并发的结果集合并在一起才是完整的结果集。
- 每次返回行数
limit默认为2000，最大可以到2000。超过2000后，limit的变化对性能基本无影响。
- 性能
ParallelScan接口单并发扫描数据的性能是Search接口的5倍。当增大并发数时，性能可以继续线性提高，例如8并发时仍可以继续提高4倍性能。
- 成本
由于ParallelScan请求对资源的消耗更少，价格会更便宜，所以对于大数据量的导出类需求，强烈建议使用ParallelScan接口。
- 返回列
只支持返回多元索引中已建立索引的列，不能返回多元索引中没有的列，可以支持的ReturnType是RETURN_ALL_INDEX或者RETURN_SPECIFIED，不支持RETURN_ALL。
目前已支持返回数组和地理位置，但是返回的字段值会被格式化，可能和写入数据表的值不一致。例如对于数组类型，写入数据表的值为"[1,2, 3, 4]"，则通过ParallelScan接口导出的值为"[1,2,3,4]"；对于地理位置类型，写入数据表的值为
10,50，则通过ParallelScan接口导出的值为10.0,50.0。 - 限制项
同时存在的ParallelScan任务数量有一定的限制，当前为10个，后续会根据客户需求继续调整。同一个sessionId且ScanQuery相同的多个并发任务视为一个任务。一个ParallelScan任务的生命周期定义为：开始时间是第一次发出ParallelScan请求，结束时间是翻页扫描完所有数据或者请求的token失效。
接口
多并发数据导出功能涉及如下两个接口。
- ComputeSplits：获取当前ParallelScan单个请求的最大并发数。
- ParallelScan：执行具体的数据导出功能。
使用
您可以使用如下语言的SDK并发导出数据。
参数
| 参数 | 说明 | |
|---|---|---|
| tableName | 数据表名称。 | |
| indexName | 多元索引名称。 | |
| scanQuery | query | 多元索引的查询语句。支持精确查询、模糊查询、范围查询、地理位置查询、嵌套查询等，功能和Search接口一致。 |
| limit | 扫描数据时一次能返回的数据行数。 | |
| maxParallel | 最大并发数。请求支持的最大并发数由用户数据量决定。数据量越大，支持的并发数越多，每次任务前可以通过ComputeSplits API进行获取。 | |
| currentParallelId | 当前并发ID。取值范围为[0, maxParallel)。 | |
| token | 用于翻页功能。ParallelScan请求结果中有下一次进行翻页的token，使用该token可以接着上一次的结果继续读取数据。 | |
| aliveTime | ParallelScan的当前任务有效时间，也是token的有效时间。默认值为60，建议使用默认值，单位为秒。如果在有效时间内没有发起下一次请求，则不能继续读取数据。持续发起请求会刷新token有效时间。
说明 由于服务端采用异步方式清理过期任务，因此当前任务只保证在设置的有效时间内不会过期，但不能保证有效时间之后一定过期。
|
|
| columnsToGet | ParallelScan目前仅可以扫描多元索引中的数据，需要在创建多元索引时设置附加存储（即store=true）。 | |
| sessionId | 本次并发扫描数据任务的sessionId。创建Session可以通过ComputeSplits API来创建，同时获得本次任务支持的最大并发数。 | |
示例
单并发扫描数据和多线程并发扫描数据的代码示例如下：
- 单并发扫描数据
相对于多并发扫描数据，单并发扫描数据的代码更简单，单并发代码无需关心currentParallelId和maxParallel参数。单并发使用方式的整体吞吐比Search接口方式高，但是比多线程多并发使用方式的吞吐低，多线程多并发方式请参见最下方的“多线程并发扫描数据”示例代码。
import java.util.ArrayList; import java.util.Arrays; import java.util.List; import com.alicloud.openservices.tablestore.SyncClient; import com.alicloud.openservices.tablestore.model.ComputeSplitsRequest; import com.alicloud.openservices.tablestore.model.ComputeSplitsResponse; import com.alicloud.openservices.tablestore.model.Row; import com.alicloud.openservices.tablestore.model.SearchIndexSplitsOptions; import com.alicloud.openservices.tablestore.model.iterator.RowIterator; import com.alicloud.openservices.tablestore.model.search.ParallelScanRequest; import com.alicloud.openservices.tablestore.model.search.ParallelScanResponse; import com.alicloud.openservices.tablestore.model.search.ScanQuery; import com.alicloud.openservices.tablestore.model.search.SearchRequest.ColumnsToGet; import com.alicloud.openservices.tablestore.model.search.query.MatchAllQuery; import com.alicloud.openservices.tablestore.model.search.query.Query; import com.alicloud.openservices.tablestore.model.search.query.QueryBuilders; public class Test { public static List<Row> scanQuery(final SyncClient client) { String tableName = "<TableName>"; String indexName = "<IndexName>"; //获取sessionId和本次请求支持的最大并发数。 ComputeSplitsRequest computeSplitsRequest = new ComputeSplitsRequest(); computeSplitsRequest.setTableName(tableName); computeSplitsRequest.setSplitsOptions(new SearchIndexSplitsOptions(indexName)); ComputeSplitsResponse computeSplitsResponse = client.computeSplits(computeSplitsRequest); byte[] sessionId = computeSplitsResponse.getSessionId(); int splitsSize = computeSplitsResponse.getSplitsSize(); /* * 创建并发扫描数据请求。 */ ParallelScanRequest parallelScanRequest = new ParallelScanRequest(); parallelScanRequest.setTableName(tableName); parallelScanRequest.setIndexName(indexName); ScanQuery scanQuery = new ScanQuery(); //该query决定了扫描出的数据范围，可用于构建嵌套的复杂的query。 Query query = new MatchAllQuery(); scanQuery.setQuery(query); //设置单次请求返回的数据行数。 scanQuery.setLimit(2000); parallelScanRequest.setScanQuery(scanQuery); ColumnsToGet columnsToGet = new ColumnsToGet(); columnsToGet.setColumns(Arrays.asList("col_1", "col_2")); parallelScanRequest.setColumnsToGet(columnsToGet); parallelScanRequest.setSessionId(sessionId); /* * 使用builder模式创建并发扫描数据请求，功能与前面一致。 */ ParallelScanRequest parallelScanRequestByBuilder = ParallelScanRequest.newBuilder() .tableName(tableName) .indexName(indexName) .scanQuery(ScanQuery.newBuilder() .query(QueryBuilders.matchAll()) .limit(2000) .build()) .addColumnsToGet("col_1", "col_2") .sessionId(sessionId) .build(); List<Row> result = new ArrayList<>(); /* * 使用原生的API扫描数据。 */ { ParallelScanResponse parallelScanResponse = client.parallelScan(parallelScanRequest); //下次请求的ScanQuery的token。 byte[] nextToken = parallelScanResponse.getNextToken(); //获取数据。 List<Row> rows = parallelScanResponse.getRows(); result.addAll(rows); while (nextToken != null) { //设置token。 parallelScanRequest.getScanQuery().setToken(nextToken); //继续扫描数据。 parallelScanResponse = client.parallelScan(parallelScanRequest); //获取数据。 rows = parallelScanResponse.getRows(); result.addAll(rows); nextToken = parallelScanResponse.getNextToken(); } } /* * 推荐方式。 * 使用iterator方式扫描所有匹配数据。使用方式上更简单，速度和前面方法一致。 */ { RowIterator iterator = client.createParallelScanIterator(parallelScanRequestByBuilder); while (iterator.hasNext()) { Row row = iterator.next(); result.add(row); //获取具体的值。 String col_1 = row.getLatestColumn("col_1").getValue().asString(); long col_2 = row.getLatestColumn("col_2").getValue().asLong(); } } /* * 关于失败重试的问题，如果本函数的外部调用者有重试机制或者不需要考虑失败重试问题，可以忽略此部分内容。 * 为了保证可用性，遇到任何异常均推荐进行任务级别的重试，重新开始一个新的ParallelScan任务。 * 异常分为如下两种： * 1、服务端Session异常OTSSessionExpired。 * 2、调用者客户端网络等异常。 */ try { //正常处理逻辑。 { RowIterator iterator = client.createParallelScanIterator(parallelScanRequestByBuilder); while (iterator.hasNext()) { Row row = iterator.next(); //处理row，内存足够大时可直接放到list中。 result.add(row); } } } catch (Exception ex) { //重试。 { result.clear(); RowIterator iterator = client.createParallelScanIterator(parallelScanRequestByBuilder); while (iterator.hasNext()) { Row row = iterator.next(); //处理row，内存足够大时可直接放到list中。 result.add(row); } } } return result; } } - 多线程并发扫描数据
import com.alicloud.openservices.tablestore.SyncClient; import com.alicloud.openservices.tablestore.model.ComputeSplitsRequest; import com.alicloud.openservices.tablestore.model.ComputeSplitsResponse; import com.alicloud.openservices.tablestore.model.Row; import com.alicloud.openservices.tablestore.model.SearchIndexSplitsOptions; import com.alicloud.openservices.tablestore.model.iterator.RowIterator; import com.alicloud.openservices.tablestore.model.search.ParallelScanRequest; import com.alicloud.openservices.tablestore.model.search.ScanQuery; import com.alicloud.openservices.tablestore.model.search.query.QueryBuilders; import java.util.ArrayList; import java.util.List; import java.util.concurrent.Semaphore; import java.util.concurrent.atomic.AtomicLong; public class Test { public static void scanQueryWithMultiThread(final SyncClient client, String tableName, String indexName) throws InterruptedException { // 获取机器的CPU数量。 final int cpuProcessors = Runtime.getRuntime().availableProcessors(); // 指定客户端多线程的并发数量。建议和客户端的CPU核数一致，避免客户端压力太大，影响查询性能。 final Semaphore semaphore = new Semaphore(cpuProcessors); // 获取sessionId和本次请求支持的最大并发数。 ComputeSplitsRequest computeSplitsRequest = new ComputeSplitsRequest(); computeSplitsRequest.setTableName(tableName); computeSplitsRequest.setSplitsOptions(new SearchIndexSplitsOptions(indexName)); ComputeSplitsResponse computeSplitsResponse = client.computeSplits(computeSplitsRequest); final byte[] sessionId = computeSplitsResponse.getSessionId(); final int maxParallel = computeSplitsResponse.getSplitsSize(); // 业务统计行数使用。 AtomicLong rowCount = new AtomicLong(0); /* * 为了使用一个函数实现多线程功能，此处构建一个内部类继承Thread来使用多线程。 * 您也可以构建一个正常的外部类，使代码更有条理。 */ final class ThreadForScanQuery extends Thread { private final int currentParallelId; private ThreadForScanQuery(int currentParallelId) { this.currentParallelId = currentParallelId; this.setName("ThreadForScanQuery:" + maxParallel + "-" + currentParallelId); // 设置线程名称。 } @Override public void run() { System.out.println("start thread:" + this.getName()); try { // 正常处理逻辑。 { ParallelScanRequest parallelScanRequest = ParallelScanRequest.newBuilder() .tableName(tableName) .indexName(indexName) .scanQuery(ScanQuery.newBuilder() .query(QueryBuilders.range("col_long").lessThan(10_0000)) // 此处的query决定了获取什么数据。 .limit(2000) .currentParallelId(currentParallelId) .maxParallel(maxParallel) .build()) .addColumnsToGet("col_long", "col_keyword", "col_bool") // 设置要返回的多元索引中的部分字段，或者使用下行注释的内容获取多元索引中全部数据。 //.returnAllColumnsFromIndex(true) .sessionId(sessionId) .build(); // 使用Iterator形式获取所有数据。 RowIterator ltr = client.createParallelScanIterator(parallelScanRequest); long count = 0; while (ltr.hasNext()) { Row row = ltr.next(); // 增加自定义的处理逻辑，此处代码以统计行数为例介绍。 count++; } rowCount.addAndGet(count); System.out.println("thread[" + this.getName() + "] finished. this thread get rows:" + count); } } catch (Exception ex) { // 如果有异常，此处需要考虑重试正常处理逻辑。 } finally { semaphore.release(); } } } // 多个线程同时执行，currentParallelId取值范围为[0, maxParallel)。 List<ThreadForScanQuery> threadList = new ArrayList<ThreadForScanQuery>(); for (int currentParallelId = 0; currentParallelId < maxParallel; currentParallelId++) { ThreadForScanQuery thread = new ThreadForScanQuery(currentParallelId); threadList.add(thread); } // 同时启动。 for (ThreadForScanQuery thread : threadList) { // 利用semaphore限制同时启动的线程数量，避免客户端瓶颈。 semaphore.acquire(); thread.start(); } // 主线程阻塞等待所有线程完成任务。 for (ThreadForScanQuery thread : threadList) { thread.join(); } System.out.println("all thread finished! total rows:" + rowCount.get()); } }