awslabs / awslabs/amazon-dynamodb-tools
[bulk] A malformed line in an export escapes a Spark worker instead of being reported
- 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
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