Add IO template for consuming database change streams

[boetro]

Can look at starting with Google Cloud Spanner for this, but ideally we can figure out something generic enough to extend to any database that supports change streams.