7 ms·

Streaming Databases in Real-Time with Kafka, Debezium, and MySQL

110 points by mirceasoaica 10y ago

cookiecaper 10y ago
While I respect the process involved in putting so many moving parts together, sometimes I wonder why we have to complicate these things so much. From my reading of the flow chart, it appears that data jumps through 5 different data layers before it gets written in its final destination. This doesn't appear to be a transformation pipeline, that's just how much it takes to go from user input -> saved in indexed storage. What unique featureset does WePay provide that justifies this type of complication? As an industry, there's been an explosion of this massive overcomplication the last few years. We're contorting ourselves pretty bad here, and it seems primarily to be in service of the engineer's desire to, well, engineer. Can you imagine if we let road engineers get away with this kind of stuff?
luhn 10y ago
The "traditional" Kafka infrastructure (i.e. LinkedIn) uses Kafka as the source of truth. Everything gets written directly into Kafka and makes its way from there. This is a powerful concept, but involves giving up a lot of advantages of a RDBMS. This is the first time I've heard of the approach WePay uses and I really like it. Now MySQL is the source of truth and Debezium pipes it into Kafka, allowing you to leverage Kafka while still using MySQL as your primary data store. (Or Postgres, which Debezium includes support for.)
- swah 10y ago
  tl;dr but so the simple operations like "update my birthdate" happen with sync MySQL operations?
  - luhn 10y ago
    Yep. All their applications always write directly to MySQL.
    - happyslobro 10y ago
      I guess if that became a bottleneck, they could set up a queue for write requests. Maybe even something more sophisticated that automatically batches similar writes that pile up in the queue. They could use something like Kafka for this...
mslate 10y ago
Related: - https://github.com/zendesk/maxwell https://github.com/zendesk/maxwell - https://github.com/confluentinc/bottledwater-pg https://github.com/confluentinc/bottledwater-pg
- brlewis 10y ago
  Anyone know of something like maxwell, debezium or bottledwater that uses thrift instead of avro or JSON?
  - bpicolo 10y ago
    I imagine once it's in kafka you could get it from a kafka pipe in whatever format you want.
  - ewencp 10y ago
    Debezium implements Kafka Connectors, so the serialization is pluggable. Debezium can work with thrift, you just need a thrift converter for Kafka Connect.
- cookiecaper 10y ago
  Following the BottledWater-Pg link through to the Confluence blog [0], there's a hilarious illustration of what we're doing to ourselves. The first flow chart [1] is simple. It shows a user, an app, and three separated data volumes that each serve a separate use case (db, cache, and index [assuming for OLAP workloads]). The chart is headlined with the adamant imperative "Stop doing this". Instead, Confluence suggests that we start doing this [2]: user -> app -> DB -> extraction -> Kafka -> (Index <-> Cache <-> HDFS) -> monitoring -> samza. Ehhh, no thanks. I like the other option. We need to understand that good engineering is not about making more work for ourselves. It's about simplicity and elegance, and being able to accomplish complex tasks WITHOUT wrapping ourselves up into some intractable mega-contorsion. More moving parts means more fragility and more waste. Simplicity means beauty, power, and flexibility. Now, I'm not suggesting that such architectures are never justified. I just want to highlight that the complexity should be eschewed, not celebrated. If you find yourself writing a blog post that converts a simple 3-step process into a complex 5-step, 9-destination process, alarm bells should be ringing, and you should be talking about why your organization (see Conway's Law) and/or the state of computer science sucks so bad that the 3-step process isn't good enough. [0] https://www.confluent.io/blog/bottled-water-real-time-integration-of-postgresql-and-kafka/ https://www.confluent.io/blog/bottled-water-real-time-integr... [1] https://cdn2.hubspot.net/hub/540072/file-3062873213-png/blog-files/slide-37-4-3.png https://cdn2.hubspot.net/hub/540072/file-3062873213-png/blog... [2] https://cdn2.hubspot.net/hub/540072/file-3062873223-png/blog-files/slide-42b-4-3.png https://cdn2.hubspot.net/hub/540072/file-3062873223-png/blog...
barrkel 10y ago
I recently had cause to work with Kafka Connect; we needed to get data from MySQL into Hadoop. It was not a positive experience, and IMO Kafka Connect is pretty immature, and Kafka as it is currently constituted isn't well featured enough for this purpose. Kafka Connect is architected around the idea of producers and consumers that either add messages to Kafka, or read messages from Kafka. The MySQL producer isn't suited for anything other than the most basic of table replication, though; if you need any ETL, you'll be gluing more stuff into your pipeline downstream. And when the producer falls over, the first you'll know about it is when you read the logs or poll status indicators. It didn't give me warm fuzzies about reliability nor visibility nor flexibility. It was very basic stuff. The Hadoop consumer had an unpleasant surprise: you have zero choice over table name in Hive metastore; your Kafka topic will be the name of your Hive table, no ifs, no buts, no choices. And since Kafka doesn't have any namespaces, either you're going to be running multiple Kafka clusters, or you need global agreement on topics vs Hive metadata (which does have namespaces). We have a multi-tenancy architecture and use namespaces. A non-starter. Why do I think Kafka doesn't have the right feature set? Because Kafka message expiry has only two policies, as far as I could tell: time or space. Either your message is too old and gets discarded (en bloc, IIRC); or Kafka hits a space limit and starts clearing out old messages. The natural question that arises when you're using Kafka to buffer a consumer vs a producer, then, is flow control / backpressure: how do you get the producer to slow down when the consumer can't catch up? And vice versa? Well, there's knobs you can manually control to throttle the producer, but it's in your own hands. You're dancing at the edge of a cliff if a consumer has died and messages start expiring; there's nothing stopping data loss. The only way you can start to turn this situation into a win is if (a) you have such a big firehose of data that nothing else can cope, or (b) you can take advantage of network effects and use Kafka as a data bus, not just as a pipe with exactly two ends. But it has to overcome the downsides.
- Xorlev 10y ago
  > Well, there's knobs you can manually control to throttle the producer, but it's in your own hands. You're dancing at the edge of a cliff if a consumer has died and messages start expiring; there's nothing stopping data loss. At least how we run Kafka, our logs expire after 7 days, and our alerts go off pretty quickly if consumers fall behind. Additionally, we archive all our messages to S3 via a process based on Pinterest's Secor [1]. If we were to ever run so far behind that we needed to start over, we can just run mapreduce jobs to rebuild datastores and then let consumers catch back up. Since Kafka is explicitly a pub/sub replicated+partitioned log, it doesn't make sense to provide backpressure. A single ailing consumer would cascade failure through your system. If you need synchronous or bounded replication, Kafka isn't for you. Having run Kafka in production for 2 1/2 years now, I can say with certainty that we've never felt like we were lacking in terms of features from Kafka its self, nor have we ever had a consumer fall so far behind it could never catch back up. We do leverage our archives for batch jobs though. [1] https://github.com/pinterest/secor https://github.com/pinterest/secor
zzzeek 10y ago
You know I just can't shake the 90's from my architectural perspective. Back then, we tried making domain models for everything; using a relational database is great but you have it as a backing store for the domain model, and that model is your business case that's exposed to the world. The domain model is what you publish; that's what your web services see, that's what your GUIs see, and if you're publishing out replication events to go to totally other kinds of systems that aren't relational databases, that's the model you publish there as well. Back then, one of the dragons we were slaying was the "stored procedure wall on top of SQL Server" API model, where relational databases had half-broken APIs exposed as crappy SPs and views that the whole world would consume directly from an ODBC connection over many network hops. It was a great way to kill any database, and schema changes were enormous beasts since the whole world hardcoded to your schema. The article here as well as the yelp one seem to be interested in streaming out the MySQL database schema itself, not just data changes but schema changes as well. If you're doing this as part of bigger MySQL database replication scheme, e.g. that the stream is just consumed by other MySQL databases that agree to share the identical schema, that's great. But the context here seems to be more along the lines of a swarm of microservices, which we would expect are each dealing with their own private set of MySQL data, are not sending INSERT/UPDATE/DELETE statements to each other and instead are speaking more in terms of the domain model. My 90's-esque sense is that if you've built up on lots of services, the messages you want flying around the network should be in terms of the domain models of those services, not your raw SQL / DDL. Disclaimer: this is an "I'm an idiot" comment because I think I'm missing something Disclaimer: I'm the author of SQLAlchemy so yes, I get the SQL/DDL/ORM thing
- SmellTheGlove 10y ago
  I was reading your comment and getting ready to tell you that I agreed, and that SQLAlchemy helps do pretty much what you're suggesting. Then I got to the bottom. Thanks for what you do! But yeah, I might be missing something too, but by and large I've lived through a lot of code that depends on a specific database schema, rather than a domain model or some other abstraction. Even when database changes don't break things, there is so. much. regression. testing. before anyone is confident.
- ewbourget 10y ago
mattbillenstein 10y ago
I've been building some bits of this although with I think much simpler components -- our event pipeline is nginx+lua -> nsq -> python consumer (transform, etc) -> bigquery I also have nsq_to_file hanging off of here to archive both event formats - raw+bigquery It is surprisingly little code and I'm happy with its performance although we are at a smaller scale. I've never been happy with the hadoop ecosystem or java tooling around it -- it's all designed to be scalable and fault tolerant, but it seems like it's always broken if you actually run one of these things. Regarding the ETL, the thing I haven't figured out yet is what to do with the data that's not events. We do daily/hourly exports of those tables in-bulk, but it's not real-time in the data warehouse. This is mostly ok, but I'd love a magic bullet that let me stream these updates into BigQuery as well. In any case -- nice blog post -- nice to see how others are doing it.

110 points by mirceasoaica 10y ago

While I respect the process involved in putting so many moving parts together, sometimes I wonder why we have to complicate these things so much. From my reading of the flow chart, it appears that data jumps through 5 different data layers before it gets written in its final destination. This doesn't appear to be a transformation pipeline, that's just how much it takes to go from user input -> saved in indexed storage. What unique featureset does WePay provide that justifies this type of complication? As an industry, there's been an explosion of this massive overcomplication the last few years. We're contorting ourselves pretty bad here, and it seems primarily to be in service of the engineer's desire to, well, engineer. Can you imagine if we let road engineers get away with this kind of stuff?

luhn 10y ago

The "traditional" Kafka infrastructure (i.e. LinkedIn) uses Kafka as the source of truth. Everything gets written directly into Kafka and makes its way from there. This is a powerful concept, but involves giving up a lot of advantages of a RDBMS. This is the first time I've heard of the approach WePay uses and I really like it. Now MySQL is the source of truth and Debezium pipes it into Kafka, allowing you to leverage Kafka while still using MySQL as your primary data store. (Or Postgres, which Debezium includes support for.)

swah 10y ago
tl;dr but so the simple operations like "update my birthdate" happen with sync MySQL operations?
- luhn 10y ago
  Yep. All their applications always write directly to MySQL.
  - happyslobro 10y ago
    I guess if that became a bottleneck, they could set up a queue for write requests. Maybe even something more sophisticated that automatically batches similar writes that pile up in the queue. They could use something like Kafka for this...

mslate 10y ago

Related: - https://github.com/zendesk/maxwell https://github.com/zendesk/maxwell - https://github.com/confluentinc/bottledwater-pg https://github.com/confluentinc/bottledwater-pg

brlewis 10y ago
Anyone know of something like maxwell, debezium or bottledwater that uses thrift instead of avro or JSON?
- bpicolo 10y ago
  I imagine once it's in kafka you could get it from a kafka pipe in whatever format you want.
- ewencp 10y ago
  Debezium implements Kafka Connectors, so the serialization is pluggable. Debezium can work with thrift, you just need a thrift converter for Kafka Connect.
cookiecaper 10y ago
Following the BottledWater-Pg link through to the Confluence blog [0], there's a hilarious illustration of what we're doing to ourselves. The first flow chart [1] is simple. It shows a user, an app, and three separated data volumes that each serve a separate use case (db, cache, and index [assuming for OLAP workloads]). The chart is headlined with the adamant imperative "Stop doing this". Instead, Confluence suggests that we start doing this [2]: user -> app -> DB -> extraction -> Kafka -> (Index <-> Cache <-> HDFS) -> monitoring -> samza. Ehhh, no thanks. I like the other option. We need to understand that good engineering is not about making more work for ourselves. It's about simplicity and elegance, and being able to accomplish complex tasks WITHOUT wrapping ourselves up into some intractable mega-contorsion. More moving parts means more fragility and more waste. Simplicity means beauty, power, and flexibility. Now, I'm not suggesting that such architectures are never justified. I just want to highlight that the complexity should be eschewed, not celebrated. If you find yourself writing a blog post that converts a simple 3-step process into a complex 5-step, 9-destination process, alarm bells should be ringing, and you should be talking about why your organization (see Conway's Law) and/or the state of computer science sucks so bad that the 3-step process isn't good enough. [0] https://www.confluent.io/blog/bottled-water-real-time-integration-of-postgresql-and-kafka/ https://www.confluent.io/blog/bottled-water-real-time-integr... [1] https://cdn2.hubspot.net/hub/540072/file-3062873213-png/blog-files/slide-37-4-3.png https://cdn2.hubspot.net/hub/540072/file-3062873213-png/blog... [2] https://cdn2.hubspot.net/hub/540072/file-3062873223-png/blog-files/slide-42b-4-3.png https://cdn2.hubspot.net/hub/540072/file-3062873223-png/blog...

barrkel 10y ago

I recently had cause to work with Kafka Connect; we needed to get data from MySQL into Hadoop. It was not a positive experience, and IMO Kafka Connect is pretty immature, and Kafka as it is currently constituted isn't well featured enough for this purpose. Kafka Connect is architected around the idea of producers and consumers that either add messages to Kafka, or read messages from Kafka. The MySQL producer isn't suited for anything other than the most basic of table replication, though; if you need any ETL, you'll be gluing more stuff into your pipeline downstream. And when the producer falls over, the first you'll know about it is when you read the logs or poll status indicators. It didn't give me warm fuzzies about reliability nor visibility nor flexibility. It was very basic stuff. The Hadoop consumer had an unpleasant surprise: you have zero choice over table name in Hive metastore; your Kafka topic will be the name of your Hive table, no ifs, no buts, no choices. And since Kafka doesn't have any namespaces, either you're going to be running multiple Kafka clusters, or you need global agreement on topics vs Hive metadata (which does have namespaces). We have a multi-tenancy architecture and use namespaces. A non-starter. Why do I think Kafka doesn't have the right feature set? Because Kafka message expiry has only two policies, as far as I could tell: time or space. Either your message is too old and gets discarded (en bloc, IIRC); or Kafka hits a space limit and starts clearing out old messages. The natural question that arises when you're using Kafka to buffer a consumer vs a producer, then, is flow control / backpressure: how do you get the producer to slow down when the consumer can't catch up? And vice versa? Well, there's knobs you can manually control to throttle the producer, but it's in your own hands. You're dancing at the edge of a cliff if a consumer has died and messages start expiring; there's nothing stopping data loss. The only way you can start to turn this situation into a win is if (a) you have such a big firehose of data that nothing else can cope, or (b) you can take advantage of network effects and use Kafka as a data bus, not just as a pipe with exactly two ends. But it has to overcome the downsides.

Xorlev 10y ago
> Well, there's knobs you can manually control to throttle the producer, but it's in your own hands. You're dancing at the edge of a cliff if a consumer has died and messages start expiring; there's nothing stopping data loss. At least how we run Kafka, our logs expire after 7 days, and our alerts go off pretty quickly if consumers fall behind. Additionally, we archive all our messages to S3 via a process based on Pinterest's Secor [1]. If we were to ever run so far behind that we needed to start over, we can just run mapreduce jobs to rebuild datastores and then let consumers catch back up. Since Kafka is explicitly a pub/sub replicated+partitioned log, it doesn't make sense to provide backpressure. A single ailing consumer would cascade failure through your system. If you need synchronous or bounded replication, Kafka isn't for you. Having run Kafka in production for 2 1/2 years now, I can say with certainty that we've never felt like we were lacking in terms of features from Kafka its self, nor have we ever had a consumer fall so far behind it could never catch back up. We do leverage our archives for batch jobs though. [1] https://github.com/pinterest/secor https://github.com/pinterest/secor

zzzeek 10y ago

You know I just can't shake the 90's from my architectural perspective. Back then, we tried making domain models for everything; using a relational database is great but you have it as a backing store for the domain model, and that model is your business case that's exposed to the world. The domain model is what you publish; that's what your web services see, that's what your GUIs see, and if you're publishing out replication events to go to totally other kinds of systems that aren't relational databases, that's the model you publish there as well. Back then, one of the dragons we were slaying was the "stored procedure wall on top of SQL Server" API model, where relational databases had half-broken APIs exposed as crappy SPs and views that the whole world would consume directly from an ODBC connection over many network hops. It was a great way to kill any database, and schema changes were enormous beasts since the whole world hardcoded to your schema. The article here as well as the yelp one seem to be interested in streaming out the MySQL database schema itself, not just data changes but schema changes as well. If you're doing this as part of bigger MySQL database replication scheme, e.g. that the stream is just consumed by other MySQL databases that agree to share the identical schema, that's great. But the context here seems to be more along the lines of a swarm of microservices, which we would expect are each dealing with their own private set of MySQL data, are not sending INSERT/UPDATE/DELETE statements to each other and instead are speaking more in terms of the domain model. My 90's-esque sense is that if you've built up on lots of services, the messages you want flying around the network should be in terms of the domain models of those services, not your raw SQL / DDL. Disclaimer: this is an "I'm an idiot" comment because I think I'm missing something Disclaimer: I'm the author of SQLAlchemy so yes, I get the SQL/DDL/ORM thing

SmellTheGlove 10y ago
I was reading your comment and getting ready to tell you that I agreed, and that SQLAlchemy helps do pretty much what you're suggesting. Then I got to the bottom. Thanks for what you do! But yeah, I might be missing something too, but by and large I've lived through a lot of code that depends on a specific database schema, rather than a domain model or some other abstraction. Even when database changes don't break things, there is so. much. regression. testing. before anyone is confident.
ewbourget 10y ago

mattbillenstein 10y ago

I've been building some bits of this although with I think much simpler components -- our event pipeline is nginx+lua -> nsq -> python consumer (transform, etc) -> bigquery I also have nsq_to_file hanging off of here to archive both event formats - raw+bigquery It is surprisingly little code and I'm happy with its performance although we are at a smaller scale. I've never been happy with the hadoop ecosystem or java tooling around it -- it's all designed to be scalable and fault tolerant, but it seems like it's always broken if you actually run one of these things. Regarding the ETL, the thing I haven't figured out yet is what to do with the data that's not events. We do daily/hourly exports of those tables in-bulk, but it's not real-time in the data warehouse. This is mostly ok, but I'd love a magic bullet that let me stream these updates into BigQuery as well. In any case -- nice blog post -- nice to see how others are doing it.