GeezerVison

The World Wide Web, restored to 800 pixels of glory.

GeezerVison is showing github.com — perf: Parallelize DynamoDB batch reads in sync online_read by abhijeet-dhumal · Pull Request #6024 · feast-dev/feast

perf: Parallelize DynamoDB batch reads in sync online_read #6024

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged

ntkathole merged 6 commits into feast-dev:master from abhijeet-dhumal:perf/parallel-dynamodb-batch-reads

Mar 4, 2026

Uh oh!

There was an error while loading. Please reload this page.

Merged

perf: Parallelize DynamoDB batch reads in sync online_read #6024

Show file tree

Hide file tree

Changes from all commits

Commits

Show all changes

6 commits

Select commit Hold shift + click to select a range

6e65f4d

perf: Parallelize DynamoDB batch reads in sync online_read

abhijeet-dhumal Feb 25, 2026

c694ed9

test: add unit tests for DynamoDB parallel batch reads

abhijeet-dhumal Feb 25, 2026

e4a5202

fix: address thread-safety and max_workers issues in parallel DynamoD…

abhijeet-dhumal Feb 25, 2026

342587d

fix: Improve DynamoDB parallel reads: shared client, configurable wor…

abhijeet-dhumal Mar 2, 2026

2dce44b

docs: add max_read_workers config documentation for DynamoDB online s…

abhijeet-dhumal Mar 2, 2026

8ead314

docs: Fix default max worker count in docs

abhijeet-dhumal Mar 4, 2026

File filter

Filter by extension .md (1) .py (2) All 2 file types selected

Viewed files

Clear filters

Conversations

Failed to load comments.

Uh oh!

There was an error while loading. Please reload this page.

Jump to

Jump to file

Failed to load files.

Uh oh!

There was an error while loading. Please reload this page.

Diff view

Unified

Split

Hide whitespace

Diff view

Unified

Split

Hide whitespace

There are no files selected for viewing

41 changes: 41 additions & 0 deletions 41 docs/reference/online-stores/dynamodb.md

Show comments

View file

Open in desktop

Original file line number	Diff line number	Diff line change
Expand Up		@@ -22,6 +22,47 @@ online_store:

		The full set of configuration options is available in [DynamoDBOnlineStoreConfig](https://rtd.feast.dev/en/master/#feast.infra.online_stores.dynamodb.DynamoDBOnlineStoreConfig).

		## Configuration

		Below is a example with performance tuning options:

		{% code title="feature_store.yaml" %}
		```yaml
		project: my_feature_repo
		registry: data/registry.db
		provider: aws
		online_store:
		type: dynamodb
		region: us-west-2
		batch_size: 100
		max_read_workers: 10
		consistent_reads: false
		```
		{% endcode %}

		### Configuration Options

		\| Option \| Type \| Default \| Description \|
		\| ------ \| ---- \| ------- \| ----------- \|
		\| `region` \| string \| \| AWS region for DynamoDB \|
		\| `table_name_template` \| string \| `{project}.{table_name}` \| Template for table names \|
		\| `batch_size` \| int \| `100` \| Number of items per BatchGetItem/BatchWriteItem request (max 100) \|
		\| `max_read_workers` \| int \| `10` \| Maximum parallel threads for batch read operations. Higher values improve throughput for large batch reads but increase resource usage \|
		\| `consistent_reads` \| bool \| `false` \| Whether to use strongly consistent reads (higher latency, guaranteed latest data) \|
		\| `tags` \| dict \| `null` \| AWS resource tags added to each table \|
		\| `session_based_auth` \| bool \| `false` \| Use AWS session-based client authentication \|

		### Performance Tuning

		Parallel Batch Reads: When reading features for many entities, DynamoDB's BatchGetItem is limited to 100 items per request. For 500 entities, this requires 5 batch requests. The `max_read_workers` option controls how many of these batches execute in parallel:

		- Sequential (old behavior): 5 batches × 10ms = 50ms total
		- Parallel (with `max_read_workers: 10`): 5 batches in parallel ≈ 10ms total

		For high-throughput workloads with large entity counts, increase `max_read_workers` (up to 20-30) based on your DynamoDB capacity and network conditions.

		Batch Size: Increase `batch_size` up to 100 to reduce the number of API calls. However, larger batches may hit DynamoDB's 16MB response limit for tables with large feature values.

		## Permissions

		Feast requires the following permissions in order to execute commands for DynamoDB online store:
Expand Down

71 changes: 57 additions & 14 deletions 71 sdk/python/feast/infra/online_stores/dynamodb.py

Show comments

View file

Open in desktop

Original file line number	Diff line number	Diff line change
Expand Up		@@ -16,6 +16,7 @@
		import itertools
		import logging
		from collections import OrderedDict, defaultdict
		from concurrent.futures import ThreadPoolExecutor
		from datetime import datetime
		from typing import Any, Callable, Dict, List, Literal, Optional, Sequence, Tuple, Union

Expand Down Expand Up		@@ -76,6 +77,10 @@ class DynamoDBOnlineStoreConfig(FeastConfigBaseModel):
		session_based_auth: bool = False
		"""AWS session based client authentication"""

		max_read_workers: int = 10
		"""Maximum number of parallel threads for batch read operations.
		Higher values improve throughput for large batch reads but increase resource usage."""

		max_pool_connections: int = 50
		"""Max number of connections for async Dynamodb operations.
		Increase for high-throughput workloads."""
Expand Down Expand Up		@@ -479,33 +484,71 @@ def online_read(
		online_config.endpoint_url,
		online_config.session_based_auth,
		)
		table_instance = dynamodb_resource.Table(
		_get_table_name(online_config, config, table)
		)
		table_name = _get_table_name(online_config, config, table)

		batch_size = online_config.batch_size
		entity_ids = self._to_entity_ids(config, entity_keys)
		entity_ids_iter = iter(entity_ids)
		result: List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]] = []

		# Split entity_ids into batches upfront
		batches: List[List[str]] = []
		entity_ids_iter = iter(entity_ids)
		while True:
		batch = list(itertools.islice(entity_ids_iter, batch_size))

		# No more items to insert
		if len(batch) == 0:
		if not batch:
		break
		batches.append(batch)

		if not batches:
		return []

		# For single batch, no parallelization overhead needed
		if len(batches) == 1:
		batch_entity_ids = self._to_resource_batch_get_payload(
		online_config, table_instance.name, batch
		online_config, table_name, batches[0]
		)
		response = dynamodb_resource.batch_get_item(
		RequestItems=batch_entity_ids,
		response = dynamodb_resource.batch_get_item(RequestItems=batch_entity_ids)
		return self._process_batch_get_response(table_name, response, batches[0])

		# Execute batch requests in parallel for multiple batches
		# Note: boto3 clients ARE thread-safe, so we can share a single client
		# https://docs.aws.amazon.com/boto3/latest/guide/clients.html#multithreading-or-multiprocessing-with-clients
		dynamodb_client = self._get_dynamodb_client(
		online_config.region,
		online_config.endpoint_url,
		online_config.session_based_auth,
		)

		def fetch_batch(batch: List[str]) -> Dict[str, Any]:
		batch_entity_ids = self._to_client_batch_get_payload(
		online_config, table_name, batch
		)
		return dynamodb_client.batch_get_item(RequestItems=batch_entity_ids)

		# Use ThreadPoolExecutor for parallel I/O
		max_workers = min(len(batches), online_config.max_read_workers)
		with ThreadPoolExecutor(max_workers=max_workers) as executor:
		responses = list(executor.map(fetch_batch, batches))
Comment thread devin-ai-integration[bot] marked this conversation as resolved. Resolved Uh oh! There was an error while loading. Please reload this page.

		# Process responses and merge results in order
		# Client responses need deserialization (unlike resource responses)
		if self._type_deserializer is None:
		self._type_deserializer = TypeDeserializer()
		deserialize = self._type_deserializer.deserialize

		def to_tbl_resp(raw_client_response):
		return {
		"entity_id": deserialize(raw_client_response["entity_id"]),
		"event_ts": deserialize(raw_client_response["event_ts"]),
		"values": deserialize(raw_client_response["values"]),
		}

		result: List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]] = []
		for batch, response in zip(batches, responses):
		batch_result = self._process_batch_get_response(
		table_instance.name,
		response,
		batch,
		table_name, response, batch, to_tbl_response=to_tbl_resp
		)
		result.extend(batch_result)

		return result

		async def online_read_async(
Expand Down

Oops, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

Oops, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

Add this suggestion to a batch that can be applied as a single commit. This suggestion is invalid because no changes were made to the code. Suggestions cannot be applied while the pull request is closed. Suggestions cannot be applied while viewing a subset of changes. Only one suggestion per line can be applied in a batch. Add this suggestion to a batch that can be applied as a single commit. Applying suggestions on deleted lines is not supported. You must change the existing code in this line in order to create a valid suggestion. Outdated suggestions cannot be applied. This suggestion has been applied or marked resolved. Suggestions cannot be applied from pending reviews. Suggestions cannot be applied on multi-line comments. Suggestions cannot be applied while the pull request is queued to merge. Suggestion cannot be applied right now. Please check back later.

You are visitor 01402520 — GeezerVison does not store the pages you look up.
Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required.