4 ms·

All these tools built on Postgres CDC have tutorials that are based on a single table. But in real life, your data models are likely to be normalized across mul

by zknill 3y ago

All these tools built on Postgres CDC have tutorials that are based on a single table. But in real life, your data models are likely to be normalized across multiple tables with joins. Or you might have an aggregate object that contains multiple child objects from another table. The hard part of using streaming data as the data source is that you don't have a mechanism to go back and get the data that you missed. Either for some join across normalised tables or to fetch some child objects you've not seen yet. I don't see how any reasonably complicated data model would work here.

jabart 3y ago
We built a streaming CDC pipeline in MSSQL and have aggregated joins to a flat table on the analytics database side. It's like 80 lines a table with half of that being a bulk load from the beginning of time in case we ever need to add a new columns or change schemas. We built it as a command line argument though I'm sure others do something similar. 80 lines could be trimmed down if we made some base class abstraction but it shipped and we don't touch it and moved on to other features.
gunnarmorling 3y ago
One way for addressing this concern is (stateful) stream processing, for instance using Kafka Streams of Apache Flink: using a state store, you ingest the CDC streams for all the tables and incrementally update joins between them whenever there's an event coming in for either input stream. Touched on this recently in this talk about streaming data contracts: https://speakerdeck.com/gunnarmorling/data-contracts-in-practice-with-debezium-and-apache-flink-kafka-summit-london?slide=31 https://speakerdeck.com/gunnarmorling/data-contracts-in-prac.... An alternative is feeding only one CDC stream into the stream processor, typically for the root of your aggregate structure, and then re-select the entire aggregate by running a join query against the source database (discussed this here: https://www.decodable.co/blog/taxonomy-of-data-change-events https://www.decodable.co/blog/taxonomy-of-data-change-events). Both approaches have there pros and cons, e.g. in terms of required state size, (transactional) consistency, guarantees of capturing all intermediary states, etc.
- zknill 3y ago
  Of course it's possible to solve. But I think the solutions (and the cons of those solutions) are often pretty arduous or unreasonable. Take the example of joining multiple streams, you have to have all your data outside your database. You have to stream every table you want data from. And worst of all, you don't have any transactional consistency across those joins, it's possible to join when you've only consumed one half of the streaming data (e.g. one of two topics). The point is, everything seems so simple in the example but these tools often don't scale (simply) to multiple tables in the source db.
  - jgraettinger1 3y ago
    The key ingredient is leveled reads across topics. A task which is transforming across multiple joined topics, or is materializing multiple topics to an external system, needs to read across those topics in a coordinated order so that _doesn't_ happen. Minimally by using wall-clock publication time of the events so that "A before B" relationships are preserved, and ideally using transaction metadata captured from the source system(s) so that transaction boundaries propagate through the data flow. Essentially, you have to do a streaming data shuffle where you're holding back some records to let other records catch up. We've implemented this in our own platform [1] (I'm co-founder). You can also attenuate this further by deliberately holding back data from certain streams. This is handy when stream processing because you sometimes want to react when an event _doesn't_ happen. [2] We have an example that generates events when a bike-share bike hasn't moved from it's last parked station in 48 hours, for example, indicating it's probably broke [3]. It's accomplished by statefully joining a stream of rides, with that same stream but delayed 48 hours. [1] https://www.estuary.dev https://www.estuary.dev [2] https://docs.estuary.dev/concepts/derivations/#read-delay https://docs.estuary.dev/concepts/derivations/#read-delay [3] https://github.com/estuary/flow/blob/master/examples/citi-bike/idle-bikes.flow.yaml https://github.com/estuary/flow/blob/master/examples/citi-bi...
  - jerrygenser 3y ago
    Yeah the naive approach would to join on key. Otherwise can join on a transaction id and key. However it's definitely complicated to get right.
giovannibonetti 3y ago
One way to address that is by picking an analytical DB that works well with joins without needing to denormalize data. StarRocks claims [1] to do better than ClickHouse in that regard. [1] https://www.starrocks.io/blog/benchmark-test https://www.starrocks.io/blog/benchmark-test

by zknill 3y ago

jabart 3y ago

We built a streaming CDC pipeline in MSSQL and have aggregated joins to a flat table on the analytics database side. It's like 80 lines a table with half of that being a bulk load from the beginning of time in case we ever need to add a new columns or change schemas. We built it as a command line argument though I'm sure others do something similar. 80 lines could be trimmed down if we made some base class abstraction but it shipped and we don't touch it and moved on to other features.

gunnarmorling 3y ago

One way for addressing this concern is (stateful) stream processing, for instance using Kafka Streams of Apache Flink: using a state store, you ingest the CDC streams for all the tables and incrementally update joins between them whenever there's an event coming in for either input stream. Touched on this recently in this talk about streaming data contracts: https://speakerdeck.com/gunnarmorling/data-contracts-in-practice-with-debezium-and-apache-flink-kafka-summit-london?slide=31 https://speakerdeck.com/gunnarmorling/data-contracts-in-prac.... An alternative is feeding only one CDC stream into the stream processor, typically for the root of your aggregate structure, and then re-select the entire aggregate by running a join query against the source database (discussed this here: https://www.decodable.co/blog/taxonomy-of-data-change-events https://www.decodable.co/blog/taxonomy-of-data-change-events). Both approaches have there pros and cons, e.g. in terms of required state size, (transactional) consistency, guarantees of capturing all intermediary states, etc.

zknill 3y ago
Of course it's possible to solve. But I think the solutions (and the cons of those solutions) are often pretty arduous or unreasonable. Take the example of joining multiple streams, you have to have all your data outside your database. You have to stream every table you want data from. And worst of all, you don't have any transactional consistency across those joins, it's possible to join when you've only consumed one half of the streaming data (e.g. one of two topics). The point is, everything seems so simple in the example but these tools often don't scale (simply) to multiple tables in the source db.
- jgraettinger1 3y ago
  The key ingredient is leveled reads across topics. A task which is transforming across multiple joined topics, or is materializing multiple topics to an external system, needs to read across those topics in a coordinated order so that _doesn't_ happen. Minimally by using wall-clock publication time of the events so that "A before B" relationships are preserved, and ideally using transaction metadata captured from the source system(s) so that transaction boundaries propagate through the data flow. Essentially, you have to do a streaming data shuffle where you're holding back some records to let other records catch up. We've implemented this in our own platform [1] (I'm co-founder). You can also attenuate this further by deliberately holding back data from certain streams. This is handy when stream processing because you sometimes want to react when an event _doesn't_ happen. [2] We have an example that generates events when a bike-share bike hasn't moved from it's last parked station in 48 hours, for example, indicating it's probably broke [3]. It's accomplished by statefully joining a stream of rides, with that same stream but delayed 48 hours. [1] https://www.estuary.dev https://www.estuary.dev [2] https://docs.estuary.dev/concepts/derivations/#read-delay https://docs.estuary.dev/concepts/derivations/#read-delay [3] https://github.com/estuary/flow/blob/master/examples/citi-bike/idle-bikes.flow.yaml https://github.com/estuary/flow/blob/master/examples/citi-bi...
- jerrygenser 3y ago
  Yeah the naive approach would to join on key. Otherwise can join on a transaction id and key. However it's definitely complicated to get right.

giovannibonetti 3y ago

One way to address that is by picking an analytical DB that works well with joins without needing to denormalize data. StarRocks claims [1] to do better than ClickHouse in that regard. [1] https://www.starrocks.io/blog/benchmark-test https://www.starrocks.io/blog/benchmark-test