ParallelQuery

ParallelQuery

ParallelQuery(client: Connector, dry_run: bool = False)

Bases: Parallelizer

Parallel and Batch Querier for ApertureDB

This class provides the abstraction for partitioning data into batches, so that they may be processed using different threads. It is useful when you need to perform many queries or large bulk inserts/updates and want to speed up the process by executing them concurrently.

By using ParallelQuery, users can fully utilize their CPU cores and network bandwidth to communicate with ApertureDB in parallel. For information on JSON queries and batching, see the ApertureDB JSON API documentation.

Parameters:

Name	Type	Description	Default
client	Connector	The database connector.	required
dry_run	bool	Whether to run in dry run mode. Defaults to False.	False

generate_batch

generate_batch(data: List[Tuple[Commands, Blobs]]) -> Tuple[Commands, Blobs]

Here we flatten the individual queries to run them as a single query in a batch We also update the _ref values and connections refs.

Parameters:

Name	Type	Description	Default
data	list[tuple[Query, Blobs]]	The data to be batched. Each tuple contains a list of commands and a list of blobs.	required

Returns:

Name	Type	Description
commands	Commands	The batched commands.
blobs	Blobs	The batched blobs.

do_batch

do_batch(client: Connector, batch_start: int, data: List[Tuple[Commands, Blobs]]) -> None

Executes batch of queries and blobs in the database.

Parameters:

Name	Type	Description	Default
client	Connector	The database connector.	required
data	list[tuple[Commands, Blobs]]	The data to be batched. Each tuple contains a list of commands and a list of blobs.	required

It also provides a way for invoking a user defined function to handle the responses of each of the queries executed. This function can be used to process the responses from each of the corresponding queries in Parallelizer It will be called once per query, and it needs to have 4 parameters: - requests - input_blobs - responses - output_blobs Example usage:

    class MyQueries(QueryGenerator):
        def process_responses(requests, input_blobs, responses, output_blobs):
            self.requests.extend(requests)
            self.responses.extend(responses)
    loader = ParallelLoader(self.client)
    generator = MyQueries()
    loader.ingest(generator)

query

query(generator, batchsize: int = 1, numthreads: int = 4, stats: bool = False, transformers: list = None) -> None

This function takes as input the data to be executed in specified number of threads. The generator yields a tuple : (array of commands, array of blobs) Args: generator (type): The class that generates the queries to be executed. batchsize (int, optional): Number of queries per transaction. Defaults to 1. numthreads (int, optional): Number of parallel workers. Defaults to 4. stats (bool, optional): Show statistics at end of query execution. Defaults to False. transformers (list, optional): A Transformer class, a callable, or a list of Transformer classes/callables to apply to the data. Defaults to None.

debug_sample

debug_sample(**kwargs) -> None

Sample the data to be ingested for debugging purposes.