public interface QueryBatcher extends Batcher
To facilitate long-running read, update, and delete use cases, coordinates threads to process batches of uris matching a query or coming from an Iterator. Each batch of uris matching a query will come from a single forest. The host for that forest is the target of the DatabaseClient provided to the listener’s processEvent method. The query is performed directly on each forest associated with the database for the DatabaseClient provided to DataMovementManager. The end goal of each job is determined by the listeners registered with onUrisReady. The data set from which batches are made and on which processing is performed is determined by the query or Iterator used to construct this instance.
While the most custom use cases will be addressed by custom listeners, the common use cases are addressed by provided listeners, including ApplyTransformListener, DeleteListener, ExportListener, and ExportToWriterListener. The provided listeners are used by adding an instance via onUrisReady like so:
QueryBatcher qhb = dataMovementManager.newQueryBatcher(query)
.withConsistentSnapshot()
.onUrisReady( new DeleteListener() )
.onQueryFailure(exception -> exception.printStackTrace());
JobTicket ticket = dataMovementManager.startJob(qhb);
qhb.awaitCompletion();
dataMovementManager.stopJob(ticket);
Custom listeners will generally use the MarkLogic Java Client API to manipulate the documents for the uris in each batch.
QueryBatcher is designed to be highly scalable and performant. To accommodate the largest result sets, QueryBatcher paginates through matches rather than loading matches into memory. To prevent queueing too many tasks when running a query, QueryBatcher only adds another task when one completes the query and is about to send the matching uris to the onUrisReady listeners.
For pagination to succeed, you must not modify the result set during pagination. This means you must
withConsistentSnapshot(), orIterator instead of a query.Sample usage using withConsistentSnapshot():
QueryDefinition query = new StructuredQueryBuilder().collection("myCollection");
QueryBatcher qhb = dataMovementManager.newQueryBatcher(query)
.withBatchSize(1000)
.withThreadCount(20)
.withConsistentSnapshot()
.onUrisReady(batch -> {
for ( String uri : batch.getItems() ) {
if ( uri.endsWith(".txt") ) {
client.newDocumentManager().delete(uri);
}
}
})
.onQueryFailure(exception -> exception.printStackTrace());
JobTicket ticket = dataMovementManager.startJob(qhb);
qhb.awaitCompletion();
dataMovementManager.stopJob(ticket);
Example of queueing uris in memory instead of using withConsistentSnapshot():
ArrayList<String> uris = Collections.synchronizedList(new ArrayList<>());
QueryBatcher getUris = dataMovementManager.newQueryBatcher(query)
.withBatchSize(5000)
.onUrisReady( batch -> uris.addAll(Arrays.asList(batch.getItems())) )
.onQueryFailure(exception -> exception.printStackTrace());
JobTicket getUrisTicket = dataMovementManager.startJob(getUris);
getUris.awaitCompletion();
dataMovementManager.stopJob(getUrisTicket);
// now we have the uris, let's step through them
QueryBatcher performDelete = moveMgr.newQueryBatcher(uris.iterator())
.onUrisReady(new DeleteListener())
.onQueryFailure(exception -> exception.printStackTrace());
JobTicket ticket = dataMovementManager.startJob(performDelete);
performDelete.awaitCompletion();
dataMovementManager.stopJob(ticket);
To queue uris to disk (if not enough memory is available) see UrisToWriterListener.
| Modifier and Type | Method and Description |
|---|---|
boolean |
awaitCompletion()
Blocks until the job is complete.
|
boolean |
awaitCompletion(long timeout,
TimeUnit unit)
Blocks until the job is complete.
|
JobTicket |
getJobTicket()
After the job has been started, returns the JobTicket generated when the job was started.
|
QueryFailureListener[] |
getQueryFailureListeners()
Get the array of QueryFailureListener instances registered via onBatchFailure including the HostAvailabilityListener registered by default.
|
QueryBatchListener[] |
getQuerySuccessListeners()
Get the array of QueryBatchListener instances registered via onBatchSuccess.
|
boolean |
isStopped()
true if the job is terminated (last batch was finished or
DataMovementManager.stopJob was called), false otherwise |
QueryBatcher |
onQueryFailure(QueryFailureListener listener)
Add a listener to run each time there is an exception retrieving a batch of uris.
|
QueryBatcher |
onUrisReady(QueryBatchListener listener)
Add a listener to run each time a batch of uris is ready.
|
void |
retry(QueryEvent queryEvent)
Retry in the same thread to query a batch that failed.
|
void |
setQueryFailureListeners(QueryFailureListener... listeners)
Remove any existing QueryFailureListener instances registered via onBatchFailure including the HostAvailabilityListener registered by default and replace them with the provided listeners.
|
void |
setUrisReadyListeners(QueryBatchListener... listeners)
Remove any existing QueryBatchListener instances registered via onBatchSuccess and replace them with the provided listeners.
|
QueryBatcher |
withBatchSize(int batchSize)
Sets the number of uris to retrieve per batch.
|
QueryBatcher |
withConsistentSnapshot()
Specifies that matching uris should be retrieved as they were when this QueryBatcher job started.
|
QueryBatcher |
withForestConfig(ForestConfiguration forestConfig)
If the server forest configuration changes mid-job, it can be re-fetched with
DataMovementManager.readForestConfig() then set via withForestConfig. |
QueryBatcher |
withJobName(String jobName)
Sets the job name.
|
QueryBatcher |
withThreadCount(int threadCount)
Sets the number of threads added to the internal thread pool for this instance to use for retrieving or processing batches of uris.
|
getBatchSize, getForestConfig, getJobName, getThreadCountQueryBatcher onUrisReady(QueryBatchListener listener)
Add a listener to run each time a batch of uris is ready.
listener - the action which has to be done when uris are readyQueryBatcher onQueryFailure(QueryFailureListener listener)
Add a listener to run each time there is an exception retrieving a batch of uris.
These listeners will not run when an exception is thrown by a listener registered with onUrisReady. To learn more, please see Handling Exceptions in Listeners
listener - the code to run when a failure occursvoid retry(QueryEvent queryEvent)
Retry in the same thread to query a batch that failed. This method will throw an Exception if it fails again, so it can be wrapped in a try-catch block.
queryEvent - the information about the batch that failedQueryBatchListener[] getQuerySuccessListeners()
Get the array of QueryBatchListener instances registered via onBatchSuccess.
QueryFailureListener[] getQueryFailureListeners()
Get the array of QueryFailureListener instances registered via onBatchFailure including the HostAvailabilityListener registered by default.
void setUrisReadyListeners(QueryBatchListener... listeners)
Remove any existing QueryBatchListener instances registered via onBatchSuccess and replace them with the provided listeners.
listeners - the QueryBatchListener instances this batcher should usevoid setQueryFailureListeners(QueryFailureListener... listeners)
Remove any existing QueryFailureListener instances registered via onBatchFailure including the HostAvailabilityListener registered by default and replace them with the provided listeners.
listeners - the QueryFailureListener instances this batcher should useQueryBatcher withConsistentSnapshot()
Specifies that matching uris should be retrieved as they were when this QueryBatcher job started. This enables a point-in-time query so that the set of uri matches is as it was at that point in time. This requires that the server be configured to allow such queries by setting the merge timestamp to a timestamp before the job starts or a sufficiently large negative value. This should only be used when the QueryBatcher is constructed with a query, not with an Iterator. This is required when performing a delete of documents matching the query or any modification (including ApplyTransformListener) of matching documents which would cause them to no longer match the query (otherwise pagination through the result set would fail because pages shift as documents are deleted or modfied to no longer match the query).
QueryBatcher withForestConfig(ForestConfiguration forestConfig)
If the server forest configuration changes mid-job, it can be re-fetched with DataMovementManager.readForestConfig() then set via withForestConfig.
withForestConfig in interface BatcherforestConfig - the updated ForestConfigurationQueryBatcher withJobName(String jobName)
Sets the job name. Eventually, this may become useful for seeing named jobs in ops director.
withJobName in interface BatcherjobName - the name you would like to assign to this jobQueryBatcher withBatchSize(int batchSize)
Sets the number of uris to retrieve per batch. Since uris are small relative to full documents, this number should be much higher than the batch size for WriteBatcher. The default batch size is 1000.
withBatchSize in interface BatcherbatchSize - the batch size – must be 1 or greaterQueryBatcher withThreadCount(int threadCount)
Sets the number of threads added to the internal thread pool for this instance to use for retrieving or processing batches of uris. For queries these threads both retrieve and process batches. For queries one batch per forest is queued immediately, then subsequent batches per forest are only queued after each previous batch is retrieved. This means more threads than the number of forests is likely to be beneficial only when time is spent in the listeners registered with onUrisReady, for example if ApplyTransformListener, DeleteListener, ExportListener, or ExportToWriterListener are used since each of these makes additional requests to the server. For Iterators, the main thread (the one calling startJob) is used to queue all batches–so startJob will not return until all iteration is complete and all batches are queued. For Iterators this thread count is the number of threads used for processing the queued batches (running processEvent on the listeners regiested with onUrisReady).
withThreadCount in interface Batcherboolean awaitCompletion()
Blocks until the job is complete.
boolean awaitCompletion(long timeout,
TimeUnit unit)
throws InterruptedException
Blocks until the job is complete.
timeout - the maximum time to waitunit - the time unit of the timeout argumentInterruptedException - if interrupted while waitingboolean isStopped()
true if the job is terminated (last batch was finished or DataMovementManager.stopJob was called), false otherwise
isStopped in interface BatcherDataMovementManager.stopJob was called), false otherwiseJobTicket getJobTicket()
After the job has been started, returns the JobTicket generated when the job was started.
getJobTicket in interface BatcherIllegalStateException - if this job has not yet been startedCopyright © 2013-2017 MarkLogic Corporation.