Add support for snapshotting / checkpointing between pipeline stages

[boetro]

Would be nice if we offered some kind of checkpointing solution.

Imagine your process step is working fine but something was mis-configured with your sink. It would be nice if we could checkpoint things that left the processor and retry them again later instead of having to execute the entire pipeline again.