[RFC] Parallel & Batch Ingestion

[chishui]

Is your feature request related to a problem? Please describe

Problem Statements

Today, users can utilize bulk API to ingest multiple documents in a single request. All documents from this request are handled by one ingest node and on this node, if there's any ingest pipeline configured, documents are processed by pipeline one at a time in a sequential order (ref). The ingest pipeline is constituted by a collection of processors and processor is the computing unit of a pipeline. Most of the processors are pretty light weighted such as append, uppercase, lowercase, and to process multiple documents one after another or to process them in parallel would make no observable difference. But for time-consuming processors such as neural search processors, which by their nature, require more time to compute, being able to run them in parallel could save user some valuable ingest time. Apart from ingestion time, processors like neural search, can benefit from processing batch documents together as it can reduce the requests to remote ML services via batch APIs to maximally avoid hitting rate limit restriction. (Feature request: opensearch-project/ml-commons#1840, rate limit example from OpenAI: https://platform.openai.com/docs/guides/rate-limits)

Due to the lack of parallel ingestion and batch ingestion capabilities in ingest flow, we propose below solution to address them.

Describe the solution you'd like

Proposed Features

1. Batch Ingestion

An ingest pipeline is constructed by a list of processors and a single document could flow through each processor one by one before it can be stored into index. Currently, both pipeline and processor can only handle one document each time and even if with bulk API, documents are iterated and handled in sequential order. As shown in figure 1, to ingest doc1, it would firstly flow through ingest pipeline 1, then through pipeline 2. Then, the next document would go through both pipeline.

To support batch processing of documents, we'll add a batchExecute API in ingest pipeline and processors which take multiple documents as input parameters. We will provide a default implementation in Processor interface to iteratively call existingexecute API to process document one by one so that most of the processors don't need to make change and only if there's necessity for them to batch process documents (e.g. text embedding processor), they can have their own implementation, otherwise, even receiving documents altogether, they default to process them one by one.

To batch process documents, user need to use bulk API. We'll add two optional parameters for bulk API for user to enable batch feature and set batch size. Based on maximum_batch_size value, documents are split into batches.

Since in bulk API, different documents could be ingested to different indexes, indexes could use the same pipelines but in different order, e.g. index “movies” uses pipeline P1 as default pipeline, P2 as final pipeline; index “musics” uses P2 as default pipeline and P1 as final pipeline. To avoid over-complexity of handling cross indexes batching (topology sorting), we would batch documents in index level.

2. Parallel Ingestion

Apart from batch ingestion, we also propose to have parallel ingestion to accompany with batch ingestion to boost the ingestion performance. When user enables parallel ingestion, based on batch size, documents from bulk API will be split into batches, then, batches are processed in parallel with threads managed by thread pool. Although limiting the maximum concurrency of parallel ingestion, thread pool can help us protect host resources to not be exhausted by batch ingestion threads.

Ingest flow logic change

Current logic of the ingestion flow of documents can be shown from the pseudo code below:

for (document in documents) {  
    for (pipeline in pipelines) {  
        for (processor in pipeline.processors) {  
            document = processor.execute(document)  
        }  
    }  
}

We'll change the flow to logic shown below if the pipeline has enable the batch option.

if (enabledBatch) {
    batches = calculateBatches(documents);
    for (batch in batches) {
        for (pipeline in pipelines) {  
            for (processor in pipeline.processors) {  
                documents = processor.batchExecute(documents)  
            }  
        }
    }
} else if (enabledParallelBatch) {
    batches = calculateBatches(documents);
    for (batch in batches) {
        threadpool.execute(()-> {
            for (pipeline in pipelines) {  
                for (processor in pipeline.processors) {  
                    documents = processor.batchExecute(documents)  
                }  
            }
        });
    }
} else {
    // fallback to exsiting ingestion logic
}

Update to Bulk API

We propose new parameters to bulk API, all of them are optional.

Parameter	Type	Description
batch_ingestion_option	String	Configure whether to enable batch ingestion. It has three options: none, enable and parallel. By default, it's none. When set it to enable, batch ingestion is enabled, and batches are processed in sequential order. When set it to parallel, batch ingestion is enabled and batches are processes in parallel.
maximum_batch_size	Integer	The batched document size. Only work when batch ingestion option is set to enable or parallel. It's 1 by default.

3. Split and Redistribute Bulk API

Users tend to use bulk API to ingest many documents which can be very time consuming sometimes. In order to achieve lower ingestion time, they have to use multiple clients to make multiple bulk requests with smaller document size so that the requests can be distributed to different ingest nodes. To offload the burden from user side, we can support the split and redistribute work from server side and help distribute the ingest load more evenly.
Note: although brought up here, we think it's better to discuss this topic in a separate RFC doc which will be published later.

Related component

Indexing:Performance

Describe alternatives you've considered

No response

Additional context

No response

[peternied]

[Triage - attendees 1 2 3 4 5]
@chishui Thanks for creating this RFC, it looks like this could be related to [1] [2]

• [1] [RFC] Streaming Index API #3000
• [2] Streaming Index API

[chishui]

@peternied Yes, it looks like the proposed feature 3 in this RFC has very similar idea with the streaming API especially the coordinator part to load balancing the ingest load. For feature 3, it just tries to reuse the bulk API.

Feature 1 and 2 are different from streaming API as they focus on parallel and batch ingestion on a single node which would happen post streaming API or feature 3.

[msfroh]

@dbwiddis, @joshpalis -- you may be interested in this, as you've been thinking about parallel execution for search pipelines. For ingest pipelines, the use-case is a little bit more "natural", because we already do parallel execution of _bulk request (at least across shards).

@chishui, can you confirm where exactly the parallel/batch execution would run? A bulk request is received on one node (that serves as coordinator for the request), then the underlying DocWriteRequests get fanned out to the shards. Does this logic run on the coordinator or on the individual shards? I can never remember where IngestService acts.

[chishui]

@msfroh, the "parallel/batch execution" would be run on the ingest pipeline side. The DocWriteRequests are first processed by ingest pipeline and its processors on a single ingest node, then the processed documents are fanned out to shards to be indexed. To answer your question, the logic would be run on the coordinator.

[chishui]

Additional information about parallel ingestion:

Performance:

Light-weighted processors - no improvement

We benchmarked the performance on some light weighted processors (lowercase + append) with current solution and parallelized batch solution, we don't see improvement on either latency or throughput which is aligned with our expectation that they are already very fast and parallelization wouldn't help and could bring some additional overhead.

ML processors - already in async

ML processors are the processors doing heavy lifting work, but they actually put the predict logic in a thread (code) which brings the ingestion of that document to async.

Reasons to have parallel ingestion

1. A general solution: The parallel ingestion proposed here does the parallelization on document level, any time-consuming processors either existing today or introduced later can benefit from the parallelization directly without needing to make any changes.
2. Maximum concurrency: Today, if processors makes their logic async, then only itself and the following processors will be run in a separate thread, all previous processors are still run in a same thread synchronously. Parallel ingestion can make the whole ingestion flow of a document in parallel to achieve maximum concurrency.
3. Give user controls: It provides users flexibility to control concurrency level through batch size or user can even disable parallel ingestion through request parameter.
4. Less dev efforts and resource usage if other processors want to achieve concurrency: Today, if some processor wants to achieve concurrency, they have to implement their own concurrency logic and they may also need to create their own thread-pool. It's not necessary as for a single document, processor has to be run one by one and causes wasting of resources and leads to overhead when thread switching.

Reasons not to have parallel ingestion

1. There is no urgent need or immediate gain.

[model-collapse]

Scenario for batch processor in neural search document ingestion:
Since OpenSearch 2.7, ml-commons released its remote connector, allowing opensearch to connect with remote inference endpoint. However, ml-commons can take a list of strings as input but only supports to invoke the inference API on each input text one by one. The pipeline is like follows:

Intuitively, to enable the batch API of many 3rdparty LLM inference provider such as openAI and cohere, we can let ml-commons pass thru the list of strings as "a batch" to the API. Like this:

However, this kind of approach cannot fully leverage the GPU computation power because of two reasons: 1) The batch size is sticked with how many fields are being picked by the processor, but in fact each API have their suggested batch size such as 32 or 128. 2) In deep learning for NLP, text in a batch should have similar lengths in order to obtain highest GPU efficiency, but intuitively we will regard the text from different fields will have diverse length.
The best option is to implement a "batched" processor and recompose the "batches" by collecting texts from the same field. See following:

Alternative Approach
There is one approach called "Dynamic Batching" which holds flushable queues in ml-commons. Each queues will gather the text input from the requests to ml-commons with the similar lengths. When timeout or the queue is full, the queue is flushed and the batch API of inference service is invoked. The con of this approach is that ml-commons will have quite big memory consumption to hold the queues, and timeout queue's implementation is more risky (dead locks, blocking calls) than batched processors.

Why we need the batch API?
The computation model of GPU is using block-wise SIMD (single instruction with multiple data). In AI, inferencing model by stacking input tensors together (as a batch) will effectively increase the GPU utilization. This approach is a more economic choice than using single request API.

[gaobinlong]

@reta , could you also help to take a look at this RFC, thanks!

[reta]

Thanks @gaobinlong

@reta , could you also help to take a look at this RFC, thanks!

@msfroh @model-collapse @chishui I think the idea of enhancing ingest processors with batch processing is sound in general but it may have unintended consequences, due to complexity of bulk APIs in particular:

• for example, bulk support scripts and upserts, and combination of those ... changing the ingestion sequence could lead to very surprising results (by and large, bulk API has to provide some guarantees on document processing)
• also, picking up the parallelism and batching becomes a nightmare (in my opinion), just today picking the right batch for bulk is very difficult, but adding yet more parallelization / internal batching would make it much harder

Making the ingestion API streaming based (apologies again for bribing for #3000) is fundamentally a different approach to ingestion - we would be able to vary the ingestion based on how fast the documents could be ingested at this moment of time, without introducing the complexity of batch / parallelism management.

@nknize I think you mind be eager to chime in here :)