Problem with sagemaker.feature_store.feature_group.FeatureGroup.ingest()

[juliensimon]

SDK version: 2.32.0

When an ingest() worker fails to put a record, it logs an output message and continues working, e.g.:

Failed to ingest row 1147613: An error occurred (ValidationException) when calling the PutRecord operation: <output removed>
Failed to ingest row 432713: An error occurred (InternalFailure) when calling the PutRecord operation (reached max retries: 4): Internal server error. Please try again later.

Once all workers are done, the ingest() method will complete with an exception.

This creates the following problems:

1. The return value isn't populated, which makes it impossible to access the list of failed rows in IngestionManagerPandas.failed_rows

2. If the ingestion process is run in a script, a SageMaker Processing job, etc. the job is marked as failed, unless you do something like:

try:
    feature_group.ingest(data_frame=data, max_workers=max_workers, wait=True)
except Exception:
    pass

IMHO, a nicer behavior would be to catch the exception in ingest(), to return the list of failed rows identifiers, and to let the caller decide what to do next. In some use cases, ingestion errors won't be big enough of a problem to justify failing the job and any associated workflow.

[psnilesh]

@juliensimon Thanks for your feedback, and apologies for getting to this so late.

IMHO, a nicer behavior would be to catch the exception in ingest(), to return the list of failed rows identifiers, and to let the caller decide what to do next.

The main disadvantage with this approach is that some customers might forget to check the return value, and underlying job might succeed giving them false impression that all records were ingested even if some weren't.

In some use cases, ingestion errors won't be big enough of a problem to justify failing the job and any associated workflow.

You are right, but a generic SDK must treat all errors that might result in loss of data as equally bad. Customers may choose to succeed / fail processing job or any other by handling these failures in a manner appropriate to their use-case.

Handling failures is straight forward. Please refer the snippet below,

ingestion_manager =  feature_group.ingest(data_frame=data, max_workers=max_workers) # No wait=True
try:
   ingestion_manager.wait()  # Has the same effect as wait=True on ingest(..) method. 
except IngestionError:
   failed_rows = ingestion_manager.failed_rows
   # Decide if the failures are critical enough to fail the processing job and handle it accordingly. 

[martinRenou]

Closing as answered with the following solution:

ingestion_manager =  feature_group.ingest(data_frame=data, max_workers=max_workers) # No wait=True
try:
   ingestion_manager.wait()  # Has the same effect as wait=True on ingest(..) method. 
except IngestionError:
   failed_rows = ingestion_manager.failed_rows
   # Decide if the failures are critical enough to fail the processing job and handle it accordingly