awslabs / awslabs/amazon-dynamodb-tools

[bulk] A malformed line in an export escapes a Spark worker instead of being reported

Open
#337 0 comments 0 reactions 0 assignees View on GitHub
bulk_executor help wanted
Dominant language
Python
Stars
181
Forks
50
Avg merge
4h 11m
Merged PRs (30d)
36

Description

`read_and_parse` builds `records_rdd = all_lines_rdd.map(parser.parse_to_record)`, and the parsers raise on a bad line:

```python
# full_export_parser.parse_to_record (incremental_export_parser is the same shape)
raise ValueError(f"Malformed JSON: {e}")
raise ValueError("Export line must be a JSON object")
raise ValueError("Export line missing 'Item' field")
```

Nothing catches them, so a single bad line escapes a **worker**: four Spark task retries, the job aborts, and the driver sees a Py4J wrapper with the cause buried — the #327 shape, still live in the export path. Seven raises across the two parsers.

Note that changing the exception type alone fixes nothing: a `BulkExecutorError` escaping a worker escapes just the same. It has to be caught and recorded, the way `_apply_transform` and `_resolve_and_validate` already do in `shared/export/pipeline/__init__.py`.

## The fix

Thread an error accumulator into `read_and_parse`, wrap the parse in a closure that catches, records via `record_understood_failure` (a malformed export file is the user's data, not our bug), and returns `None` for the filter to drop. `writer.py` already calls `raise_first_worker_error`, so the job then fails with the one-line reason.

## Decided: one bad line aborts the load

A malformed line in a DynamoDB export shouldn't happen, so it is not something to tolerate and count — the job should fail and say why.

One consequence to state in the message rather than engineer around: Spark evaluates the `map` lazily, so the bad line is discovered while other partitions are already writing. Aborting therefore means some items are already in the table. Avoiding that would need a full pre-pass over every line before any write, doubling the read of the export, which is not worth it for a case that shouldn't occur.

## Not in scope of #258

#258 converts the driver-side export checks (validators, `FileLoader.parse_s3_path`, `ParserFactory.get_parser`) to `BulkExecutorError` and closes #254. This is worker-side plumbing plus the abort semantics above, so it is its own change.

Contributor guide

Open the contributing guide

Research direction

Start with read_and_parse and the parse_to_record methods in full_export_parser and incremental_export_parser. Compare their error handling with _apply_transform and _resolve_and_validate in shared/export/pipeline/__init__.py, then inspect writer.py and raise_first_worker_error. Done means malformed lines are recorded and filtered, while the load aborts with the reported one-line reason.

Written by the indexing model from the issue text.

Assessment

Tech stack
python
Domain
data-engineering
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Active
Clarity
Clearly specified
Newbie friendliness
68/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.