Skip to content

feat: Improvements to Lambda materialization engine #5379

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.

Sign up for GitHub

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
franciscojavierarceo merged 5 commits into from
Jun 1, 2025
Merged

feat: Improvements to Lambda materialization engine #5379

Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
85131e4
Set read_timeout for lambda client
Mar 28, 2025
276f8db
Handle empty return from offline store
Apr 30, 2025
7a217d2
Add lambda read timeout retries
Apr 30, 2025
9bb2f0e
Add error handling in Lambda materialization engine
May 19, 2025
e32b844
Fix error handling
May 20, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Add lambda read timeout retries 
Signed-off-by: hkuepers <hanno.kuepers@ratepay.com>
  • Loading branch information
hkuepers committed May 21, 2025
commit 7a217d25b74ce655001cc2a75658433428a1072b
20 changes: 19 additions & 1 deletion sdk/python/feast/infra/materialization/aws_lambda/lambda_engine.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@

DEFAULT_BATCH_SIZE = 10_000
DEFAULT_TIMEOUT = 600
LAMBDA_TIMEOUT_RETRIES = 5

logger = logging.getLogger(__name__)

Expand Down Expand Up @@ -225,7 +226,7 @@ def _materialize_one(
logger.info("Invoking materialization for %s", path)
futures.append(
executor.submit(
self.lambda_client.invoke,
self.invoke_with_retries,
FunctionName=self.lambda_name,
InvocationType="RequestResponse",
Payload=json.dumps(payload),
Expand Down Expand Up @@ -253,3 +254,20 @@ def _materialize_one(
if not not_done
else MaterializationJobStatus.ERROR,
)

def invoke_with_retries(self, **kwargs):
"""Invoke the Lambda function and retry if it times out.

The Lambda function may time out initially if many values are updated
and DynamoDB throttles requests. As soon as the DynamoDB tables
are scaled up, the Lambda function can succeed upon retry with higher
throughput.

"""
retries = 0
while retries < LAMBDA_TIMEOUT_RETRIES:
response = self.lambda_client.invoke(**kwargs)
if "Task timed out after" not in json.loads(response["Payload"].read()):
break
retries += 1
return response