I think I've finally got it working.

Here's the trick, the one thing that was needed to solve it all: wrap the
command pipeline in a Future block, then use an Await on it with a timeout.
This way, when the parser stalls up (which it does on certain inputs), the
input is just skipped. This means that after like 30 seconds the task will go
on to read the next record, preventing it from getting killed by Hadoop.