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

[hunterhacker]

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

# 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.