11 ms·

Delivering Billions of Messages Exactly Once

492 points by fouadmatin 9y ago

throwaway67 9y ago
... or they could have used BigQuery with a primary key on message ID.
- redmalang 9y ago
  BQ doesn't have primary keys. Perhaps you are thinking of the id that can be supplied with the streaming insert? This has very loose guarantees on what is de-duplicated (~5m iirc)
  - vgt 9y ago
    yea I think within the context of BigQuery the most sensible thing would be to do an aggregate per the column that would be considered a primary key. For example [0]. That said, Streaming API de-dupe window is very nice in practice. I mentioned elsewhere on Google Cloud the most elegant way of doing this is with Google Cloud Dataflow [1] (work at G) [0]https://stackoverflow.com/questions/38446499/bigquery-deduplication-on-two-columns-as-unique-key https://stackoverflow.com/questions/38446499/bigquery-dedupl... [1]https://cloud.google.com/blog/big-data/2017/06/how-qubit-deduplicates-streaming-data-at-scale-with-google-cloud-platform https://cloud.google.com/blog/big-data/2017/06/how-qubit-ded...
stratosgear 9y ago
Site seems to be down. Any ideas how big these HN hugs of death usually are? How big of a traffic spike brings these servers down?
- Exuma 9y ago
  It's not down
  - stratosgear 9y ago
    Hmm weird. I get: This site can’t be reached segment.com refused to connect. Try: Checking the connection ERR_CONNECTION_REFUSED Maybe something local to me only?
    - psadauskas 9y ago
      Its probably an ad- or tracker-blocker. I had to disable mine to be able to load segment.com, since they're pretty much a tracking company.
    - fouadmatin 9y ago
      Might be an adblock issue? (some blockers treat direct access to a website and loading an anlytics snippet as the same)
    - asadlionpk 9y ago
      You are probably blocking this site in your hosts file. Since it's a tracking tool.
    - disconnected 9y ago
      Check your ad blocker/hosts file. In here, uMatrix just blocks the site with the message: > uMatrix has prevented the following page from loading: > https://segment.com/blog/exactly-once-delivery/ https://segment.com/blog/exactly-once-delivery/ I checked, and one of my uMatrix hosts files includes 'www.segment.com'.
      - stratosgear 9y ago
        Yep. Disabling AdAway seems to do the trick. Thanks for the heads up
        goodplay 9y ago
        You can also add an exception for a blocked site by marking it green and clicking on the opened padlock in the umatrix panel. No need to nuke a city to get a fly.
- fouadmatin 9y ago
  hm seems to be up for me? you might've caught our autoscaler mid-scaling!
- crgwbr 9y ago
  I got front paged once a few years ago, it was about 15,000 page views in the course of a few hours. It's probably grown since then, but sure by what factor.
ggcampinho 9y ago
Isn't the new feature of Kafka about this? https://issues.apache.org/jira/browse/KAFKA-4815 https://issues.apache.org/jira/browse/KAFKA-4815
- caust1c 9y ago
  It would help, but messages are sent to the API first. We aren't sending messages directly to Kafka from the Internet, of course. So we can get duplicate API submissions regardless of whether or not we enabled transactional productions into kafka from a producer.
StreamBright 9y ago
"The single requirement of all data pipelines is that they cannot lose data." Unless the business value of data is derived after applying some summary statistics, than even sampling the data works, and you can lose events in an event stream, while not changing the insight gained. Originally Kafka was designed to be a high throughput data bus for analytical pipeline where losing messages was ok. More recently they are experimenting with exactly once delivery.
- vannevar 9y ago
  Yeah, this was a major overstatement. There are lots of data pipelines where it's ok to lose some data. Consider a sensor that sends measurements hundreds of time a second to an app that operates on a 1-second timeframe. And UDP is used all the time on the internet, yet carries no delivery guarantee.
wonderwonder 9y ago
Would something like AWS SQS not scale for something like this? We currently push about 25k daily transactions over SQS, obviously no where near the scale of this, just wondering about what limitations we will bump into potentially.
- psadauskas 9y ago
  SQS is not "exactly once", so might not meet their requirements.
  - idunno246 9y ago
    sqs has fifo queues which claim to be exactly once
    - lostcolony 9y ago
      SQS queues deduplicate over a 5 minute window. This is claiming a much larger window. Either way your listener(s) still has to have its own deduplication. Ensuring a message ends up on the queue only once, and ensuring it's processed exactly once, are two different problems that require separate handling (and, the former is what most out of the box systems claim to solve, while the latter is more important, and, frankly, completely negates the need of the former).
  - twodave 9y ago
    Actually, it can be. You just pay more.
- eropple 9y ago
  (edit: incorrect, my bad, see thread)
  - twodave 9y ago
    This is wrong. There is a type of SQS queue that indeed does exactly-once. It costs more, and not the default option, but it is there.
ju-st 9y ago
52 requests, 5.4 MB and 8.63 seconds to load a simple blog post. With a bonus XHR request every 5 seconds.
- heipei 9y ago
  That's not everything: This website contacted 56 IPs in 6 countries across 47 domains to perform 100 HTTP transactions. In total, 3 MB of data was transfered, which is 5 MB uncompressed. It took 4.103 seconds to load this page. 37 cookies were set, and 8 messages to the console were logged. https://urlscan.io/result/b2e27a08-1298-491a-863f-8cadc45e73eb https://urlscan.io/result/b2e27a08-1298-491a-863f-8cadc45e73...
- joshuagvk 9y ago
  And each one made only once!
- numbsafari 9y ago
  ... and this is why friends don't let friends use Segment.
iampims 9y ago
If the OP doesn't mind expanding a little on this bit, I'd be grateful. > If the dedupe worker crashes for any reason or encounters an error from Kafka, when it re-starts it will first consult the “source of truth” for whether an event was published: the output topic. Does this mean that "on worker crash" the worker replays the entire output topic and compare it to the rocksdb dataset? Also, how do you handle scaling up or down the number of workers/partitions?
- ryanworl 9y ago
  I'm not the OP, but changing the number of Kafka partitions isn't a super graceful operation. You would be wise to add as many as you could reasonably need assuming one consumer thread per partition. But not too many because each one is at least two files on disk!
  - tedmiston 9y ago
    I haven't used Kafka yet, but a Kafka partition is roughly the same as a Kinesis shard, right?
    - ryanworl 9y ago
      Yes, but you have to set them up upfront and there is no API to split and merge partitions. If you add more partitions, it doesn't automatically re-shard for you either. I don't think it should automatically re-shard as it would cause a ton of disk and network IO, but just something to be aware of.
robotresearcher 9y ago
> [I]t’s pretty much impossible to have messages only ever be delivered once. IIRC, it's provably impossible in a distributed system where processes might fail, i.e. all real systems.
mamon 9y ago
"Exactly once" model of message is theoretically impossible to do in distributed environment with nonzero possibility of failure. If you haven't received acknowledgement from the other side of communication in the specified amount of time you can only do one of two things: 1) do nothing, risking message loss 2) retransmit, risking duplication But of course that's only from messaging system point of view. Deduplication at receiver end can help reduce problem, but itself can fail (there is no foolproof way of implementing that pseudocode's "has_seen(message.id)" method)
- malandrew 9y ago
  An overview of FLP Impossibility http://the-paper-trail.org/blog/a-brief-tour-of-flp-impossibility/ http://the-paper-trail.org/blog/a-brief-tour-of-flp-impossib...
- LgWoodenBadger 9y ago
  I agree. I wish more messaging platforms would recognize this and stop trying to paper-over the failure mode (Kafka just came out with "Exactly Once," which I think is a variant of the 2-Phase-Commit protocol, which still does not solve the problem). My go-to for explaining to people is the Two Generals Problem https://en.wikipedia.org/wiki/Two_Generals%27_Problem https://en.wikipedia.org/wiki/Two_Generals%27_Problem
  - LgWoodenBadger 9y ago
    After watching the video from the Kafka Summit talk https://www.confluent.io/kafka-summit-nyc17/introducing-exactly-once-semantics-in-apache-kafka/ https://www.confluent.io/kafka-summit-nyc17/introducing-exac... it's an essentially-useless guarantee for any sort of Kafka consumer that interacts with an external system. Exactly-Once is not something that can be provided-for or delegated-to an arbitrary, non-integrated system. For an "integrated" system, it requires idempotent behavior, in which case it's really At-Least-Once, so...
- ecesena 9y ago
  I think "exactly once" doesn't imply "non-reprocessing" (sorry for the double negative). Meaning, you want "exactly once" and you don't want duplicates, yes. But you allow for reprocessing, provided that you have a way for deduplicating. You want a guarantee that if the producer (at the top of your data processing pipeline) sends a message, then this message eventually corresponds to exactly 1 record in your final storage(s). One easy-to-understand-yet-simplistic example is: send a message to kafka, use topic+partition+offset as primary key, store in a RDBMS. This is widely accepted as "exactly once", but clearly you may have multiple attempts to save the message into the db, which will fail due to the primary key integrity constrain.
mooneater 9y ago
Awesome story. What I would like to hear more about, is the people side. The teams and personalities involved with coming up with this new system and the transition.
falcolas 9y ago
So, a combination of a best effort "at least once" messaging with deduplication near the receiving edge. Fairly standard, honestly. There is still a potential for problems in the message delivery to the endpoints (malformed messages, Kafka errors, messages not being consumed fast enough and lost), or duplication at that level (restart a listener on the Kafka stream with the wrong message ID) as well. This is based on my own pains with Kinesis and Lambda (which, I know, isn't Kafka). In my experience, better to just allow raw "at least once" messaging and perform idempotant actions based off the messages. It's not always possible (and harder when it is possible), but its tradeoffs mean you're less likely to lose messages.
- caust1c 9y ago
  This is generally better, but we're delivering these messages to integrations which don't necessarily take idempotent actions.
incan1275 9y ago
To be fair, they are upfront in the beginning about not being able to adhere to an exactly-once model. "In the past three months we’ve built an entirely new de-duplication system to get as close as possible to exactly-once delivery" What's annoying is that they do not get precise and formal about what they want out of their new model. Also, their numbers only speak to performance, not correctness. On the plus side, I think it's awesome to see bloom filters successfully used in production. That sort of thing is easy to implement, but not easy to get right for every use case.
qsymmachus 9y ago
It's funny, at my company we implemented deduplication almost exactly the same way for our push notification sender. The scale is smaller (about 10k rpm), but the basic idea is the same (store a message ID in a key-value store after each successful send). I like the idea of invalidating records by overall size, we hadn't thought of that. We just use a fixed 24-hour TTL.
bmsatierf 9y ago
In terms of connectivity, we deal with a similar problem here at CloudWalk to process payment transactions from POS terminals, where most of them rely on GPRS connections. Our network issues are nearly 6 times higher (~3.5%) due to GPRS, and we solved the duplication problem with an approach involving both client and server side. Clients would always ensure that all the information sent by the server was successfully received. If something goes wrong, instead of retrying (sending the payment again), the client sends just the transaction UUID to the server, and the server might either respond with: A. the corresponding response for the transaction or B. not found. In the scenario A, the POS terminal managed to properly send all the information to the server but failed to receive the response. In the scenario B, the POS terminal didn't even manage to properly send the information to the server, so the POS can safely retry.
- goldenkey 9y ago
  lòok maah no race cawnditions!!
- v1shnu 9y ago
  Why not just send the data too along with the UUID? It'll save another roundtrip in case of scenario B right? Or do you have data to prove that scenario B is a lot less likely to occur, making it sensible to save bandwidth by not re-transmitting the data?
linkmotif 9y ago
It's worth noting that the next major Kafka release (0.11, out soon) will include exactly once semantics! With basically no configuration and no code changes for the user. Perhaps even more noteworthy is this feature is built on top of a new transactions feature [0]. With this release, you'll be able to atomically write to multiple topics. [0] https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+E...
squiguy7 9y ago
I wonder how they partition by "messageID" they use to ensure that the de-duplication happens on the same worker. I would imagine that this affects their ability to add more brokers in the future. Perhaps they expect a 1:1 mapping of RocksDB, partition, and de-duplication worker.
- squeaky-clean 9y ago
  I'm also curious about this. I'm guessing it's some function on the id that maps it to a partition. What happens when you add more consumers? Is there a way to know which partitions include messageIds that would be put into another partition's ownership by the change and then move anything from one RocksDB instance to the other?
- bytecodes 9y ago
  Kafka does this as part of its design. A topic has a declared number of partitions (which can't really be changed on the fly, you choose a theoretical high number and hope it's enough), and an agreed upon hash algorithm chooses between those partitions (probably in Java, so hashCode is readily available for primitives as well as objects). Each partition is really like its own topic, so you lose in-order messaging for anything not included in your partition key.
gsmethells 9y ago
Why do I get the feeling this is repeating TCP features at the Message level? There must a protocol that can hide this exactly once need away. TCP doesn't create downloads, generally, that are bad and fail their checksum test, hence packets that make up the file are not duplicated.
- jmaygarden 9y ago
  How would you use TCP sockets to de-duplicate Kafka streams with a many-to-many communication pattern? Surely there is a valid scalability reason for why AWS IoT only provides "at least once" guarantees in their MQTT broker even when TCP is the underlying transport [1]. [1] http://docs.aws.amazon.com/iot/latest/developerguide/protocols.html#mqtt http://docs.aws.amazon.com/iot/latest/developerguide/protoco...
- jkarneges 9y ago
  Yes there is some duplication of TCP capability here. The problem with relying on TCP for reliability is that its state is in memory, associated with a particular peer IP address, and acknowledgements passed back to the sender only indicate that the receiver has the data in local memory, not that the data has been processed. A file download over TCP can fail, for example due to a network problem. Ensuring reliable delivery requires additional measures outside of TCP, such as retrying the download using a new connection. In practice, this means that TCP is primarily useful for providing flow control and offering a streaming interface (no worry about packet sizes). Less so as a complete solution for transmission reliability.
kortox 9y ago
With deduplication state on the worker nodes, how does scaling up, or provisioning new machines, or moving a partition between machines work?
PinguTS 9y ago
That reminds me of the safety-related protocols we use since years in embedded electronics like rail-road signaling, medical, and other areas.
spullara 9y ago
This isn't the solution I would architect. It is much easier to de-duplicate when processing your analytics workload later and you don't need to do so much work.
- tejasmanohar 9y ago
  That's not true for all the hundreds of real-time integrations that Segment sends data to. Many are write-only.
  - spullara 9y ago
    Fair. For that case this solution makes more sense than the pure analytics case.
newobj 9y ago
I don't want to ever see the phrase "Exactly Once" without several asterisks behind it. It might be exactly once from an "overall" point of view, but the client effectively needs infinitely durable infinite memory to perform the "distributed transaction" of acting on the message and responding to the server. Imagine: - Server delivers message M - Client process event E entailed by message M - Client tries to ack (A) message on server, but "packet loss" - To make matters worse, let's say the client also immediately dies after this How do you handle this situation? The client must transactionally/simultaneously commit both E and A/intent-to-A. Since the server never received an acknowledgment of M, it will either redeliver the message, in which case some record of E must be kept to deduplicate on, or it will wait for client to resend A, or some mixture of both. Note: if you say "just make E idempotent", then you don't need exactly-once delivery in the first place... I suppose you could go back to some kind of lock-step processing of messages to avoid needing to record all (E,A) that are in flight, but that would obviously kill throughput of the message queue. Exactly Once can only ever be At Least Once with some out-of-the-box idempotency that may not be as cheap as the natural idempotency of your system. EDIT: Recommended reading: "Life Beyond Distributed Transactions", Pat Helland - http://queue.acm.org/detail.cfm?id=3025012 http://queue.acm.org/detail.cfm?id=3025012
- Cieplak 9y ago
  Basically the Two General's Problem, eh? https://en.wikipedia.org/wiki/Two_Generals%27_Problem https://en.wikipedia.org/wiki/Two_Generals%27_Problem
  - newobj 9y ago
    Exactly.
    - jancsika 9y ago
      Question: what is the relationship (if any) between TCP and the Two Generals Problem?
      - 9y ago
travisjeffery 9y ago
Kafka 0.11 (recently released) has exactly once semantics and transactional messages built-in. - Talk from Kafka Summit: https://www.confluent.io/kafka-summit-nyc17/resource/#exactly-once-semantics_slide https://www.confluent.io/kafka-summit-nyc17/resource/#exactl... - Proposal: https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+E...
- mshenfield 9y ago
  My understanding (noob here) is that it allows producers to retry without fear of duplication. You still have to consider the system feeding the producer though. In Segment's example, clients might deliver their messages more than once to the API. Kafka's mechanism wouldn't detect duplicate messages sent to the producer, just that any given message a producer wants to append to Kafka won't be duplicated.
sethammons 9y ago
"Exactly Once" Over a window of time that changes depending on the amount of ingested events. Basically, they read from a kafka stream and have a deduplication layer in rocks db that produces to another kafka stream. They process about 2.2 billion events through it per day. While this will reduce duplicates and get closer to Exactly Once (helping reduce the two generals problem on incoming requests and potentially work inside their data center), they still have to face the same problem again when they push data out to their partners. Some packet loss, and they will be sending out duplicate to the partner. Not to downplay what they have done as we are doing a similar thing near our exit nodes to do our best to prevent duplicate events making it out of our system.
subhsandhu1 9y ago
Hawa Hawa Lyrics Mubarakan: http://www.desilyrics.in/2017/06/hawa-hawa-lyrics-mubarakan.html http://www.desilyrics.in/2017/06/hawa-hawa-lyrics-mubarakan....
ratherbefuddled 9y ago
"Almost Exactly Once" doesn't have quite the same ring to it, but it is actually accurate. We've already discovered better trade-offs haven't we?
majidazimi 9y ago
What is so exciting about this? There is still possibility of duplicates. You still have to put the engineering effort to deal with duplicates end-to-end. If the code is there to deal with duplicates end-to-end, then does it really matter to have 5 duplicates or 35? Or may be they just did it to add some useful cool-tech in to CV?
- ngrilly 9y ago
  > There is still possibility of duplicates. Where?
deleted 9y ago
[deleted]
vgt 9y ago
Qubit's strategy to do this via streaming, leveraging Google Cloud Dataflow: https://cloud.google.com/blog/big-data/2017/06/how-qubit-deduplicates-streaming-data-at-scale-with-google-cloud-platform https://cloud.google.com/blog/big-data/2017/06/how-qubit-ded...
redmalang 9y ago
Another possible approach: https://cloud.google.com/blog/big-data/2017/06/how-qubit-deduplicates-streaming-data-at-scale-with-google-cloud-platform https://cloud.google.com/blog/big-data/2017/06/how-qubit-ded...
siliconc0w 9y ago
Was thinking a 'reverse bloom filter' could be cool to possibly avoid the RocksDB for situations like this- turns out it already exists: https://github.com/jmhodges/opposite_of_a_bloom_filter https://github.com/jmhodges/opposite_of_a_bloom_filter I love it when that happens.
- openasocket 9y ago
  It should be noted it's impossible to have a 'reverse bloom filter' with the same properties as a regular bloom filter. That is, a constant predefined size gives you a particular false negative rate. The thing you linked to is really just a cache. It has to store an entire entry (not just set some bits based on the hash) and doesn't have a predictable false negative rate based on its size. For more info: https://cstheory.stackexchange.com/questions/6596/a-probabilistic-set-with-no-false-positives https://cstheory.stackexchange.com/questions/6596/a-probabil...
jkestelyn 9y ago
Relevant to this topic: Description of exactly-once implementation in Google Cloud Dataflow + what "exactly once" means in context of streaming: https://cloud.google.com/blog/big-data/2017/05/after-lambda-exactly-once-processing-in-google-cloud-dataflow-part-1 https://cloud.google.com/blog/big-data/2017/05/after-lambda-... (Google Cloud emp speaking)
skMed 9y ago
Having built something similar with RabbitMQ in a high-volume industry, there are a lot of benefits people in this thread seem to be glossing over and are instead debating semantics. Yes, this is not "exactly once" -- there really is no such thing in a distributed system. The best you can hope for is that your edge consumers are idempotent. There is a lot of value derived from de-duping near ingress of a heavy stream such as this. You're saving downstream consumers time (money) and potential headaches. You may be in an industry where duplicates can be handled by a legacy system, but it takes 5-10 minutes of manual checks and corrections by support staff. That was my exact use case and I can't count the number of times we were thankful our de-duping handled "most" cases.
luord 9y ago
This is interesting work. But I think I'll continue relying on at least once and idempotency. Exactly once is impossible anyway. > In Python (aka pseudo-pseudocode) This annoyed probably more than it should have.
openasocket 9y ago
So there's a lot of talk on here about the Two Generals Problem, so I thought I'd chime in with some misconceptions about how the Two Generals Problem relates to Exactly Once Messaging (EOM). WARNING: I'm going mostly on memory with this, I could be completely wrong. EOM is NOT strictly speaking equivalent to the Two Generals Problem, or Distributed Consensus, in an unreliable network. In distributed consensus, at some given point in time, A has to know X, A has to know B knows X, A has to know B knows A knows X, ... It has to do with the fact that the message broker is in some sense the arbitrator of truth, so the consumer(s) don't need full consensus. In an unreliable network, you can have EOM. http://ilpubs.stanford.edu:8090/483/1/2000-7.pdf http://ilpubs.stanford.edu:8090/483/1/2000-7.pdf gives some examples of how that works. HOWEVER, you can't have EOM when the consumers can fail. If a consumer fails there's no way, in general, to tell if the last message it was working on was completed. There are a couple of edge cases where you can still have EOM. For instance, a system where you have a message broker A, and a bunch of consumers that read messages x from that queue, compute f(x), and insert f(x) onto message broker B, where f(x) may be computed multiple times for the same x (i.e. if f is a pure function or you don't care about the side effects). This system can implement EOM in the presence of an unreliable network and consumer failures (I think it can handle one or both of the message brokers failing too, not 100% sure) in the sense that x will never be in broker A at the same time as f(x) is in broker B, f(x) will never be in broker B more than once for the same x, and any y in B had some x that was in A such that y = f(x).
philovivero 9y ago
tl;dr: Clickbait headline. Exactly-once delivery not even close to implemented. Typical de-duping, as you've seen and read about hundreds of times already, is what they did.
alexandercrohde 9y ago
Here's a radical solution. Instead of becoming a scala pro akka stream 200k engineer with a cluster of kafka nodes that costs your company over $100,000 of engineering time, technical debt, opportunity cost, and server costs, just put it all in bigtable, with deduping by id.... Enough of resume-driven-engineering, why does every need to reinvent the wheel?
- azernik 9y ago
  Yup. Databases, whether relational or not, have been designed to solve all these problems in a much more "bulletproof" way than your piddly [1] several-dozen-engineer team could ever manage, no matter how genius they are. [1] No disrespect meant - just a description of size. Source: running a piddly 2-person engineering team.
pfarnsworth 9y ago
Sounds very cool. A couple of questions I had: 1) What happens if they lose their rocksdb with all of the messageIds? 2) Is their kafka atleast-once delivery? How do they guarantee that kafka doesn't reject their write? Also, assuming they have set up their kafka for at least once delivery, doesn't that make the output topic susceptible to duplicates due to retries, etc? 3) >Instead of searching a central database for whether we’ve seen a key amongst hundreds of billions of messages, we’re able to narrow our search space by orders of magnitude simply by routing to the right partition. Is "orders of magnitude" really correct? Aren't you really just narrowing the search space by the number of partitions in kafka? I suppose if you have a hundred partitions, that would be 2 orders of magnitude, but it makes it sound like it's much more than that.
- ngrilly 9y ago
  > What happens if they lose their rocksdb with all of the messageIds? I'm wondering the same.

492 points by fouadmatin 9y ago

throwaway67 9y ago

... or they could have used BigQuery with a primary key on message ID.

redmalang 9y ago
BQ doesn't have primary keys. Perhaps you are thinking of the id that can be supplied with the streaming insert? This has very loose guarantees on what is de-duplicated (~5m iirc)
- vgt 9y ago
  yea I think within the context of BigQuery the most sensible thing would be to do an aggregate per the column that would be considered a primary key. For example [0]. That said, Streaming API de-dupe window is very nice in practice. I mentioned elsewhere on Google Cloud the most elegant way of doing this is with Google Cloud Dataflow [1] (work at G) [0]https://stackoverflow.com/questions/38446499/bigquery-deduplication-on-two-columns-as-unique-key https://stackoverflow.com/questions/38446499/bigquery-dedupl... [1]https://cloud.google.com/blog/big-data/2017/06/how-qubit-deduplicates-streaming-data-at-scale-with-google-cloud-platform https://cloud.google.com/blog/big-data/2017/06/how-qubit-ded...

stratosgear 9y ago

Site seems to be down. Any ideas how big these HN hugs of death usually are? How big of a traffic spike brings these servers down?

Exuma 9y ago
It's not down
- stratosgear 9y ago
  Hmm weird. I get: This site can’t be reached segment.com refused to connect. Try: Checking the connection ERR_CONNECTION_REFUSED Maybe something local to me only?
  - psadauskas 9y ago
    Its probably an ad- or tracker-blocker. I had to disable mine to be able to load segment.com, since they're pretty much a tracking company.
  - fouadmatin 9y ago
    Might be an adblock issue? (some blockers treat direct access to a website and loading an anlytics snippet as the same)
  - asadlionpk 9y ago
    You are probably blocking this site in your hosts file. Since it's a tracking tool.
  - disconnected 9y ago
    Check your ad blocker/hosts file. In here, uMatrix just blocks the site with the message: > uMatrix has prevented the following page from loading: > https://segment.com/blog/exactly-once-delivery/ https://segment.com/blog/exactly-once-delivery/ I checked, and one of my uMatrix hosts files includes 'www.segment.com'.
    - stratosgear 9y ago
      Yep. Disabling AdAway seems to do the trick. Thanks for the heads up
      - goodplay 9y ago
        You can also add an exception for a blocked site by marking it green and clicking on the opened padlock in the umatrix panel. No need to nuke a city to get a fly.
fouadmatin 9y ago
hm seems to be up for me? you might've caught our autoscaler mid-scaling!
crgwbr 9y ago
I got front paged once a few years ago, it was about 15,000 page views in the course of a few hours. It's probably grown since then, but sure by what factor.

ggcampinho 9y ago

Isn't the new feature of Kafka about this? https://issues.apache.org/jira/browse/KAFKA-4815 https://issues.apache.org/jira/browse/KAFKA-4815

caust1c 9y ago
It would help, but messages are sent to the API first. We aren't sending messages directly to Kafka from the Internet, of course. So we can get duplicate API submissions regardless of whether or not we enabled transactional productions into kafka from a producer.

StreamBright 9y ago

"The single requirement of all data pipelines is that they cannot lose data." Unless the business value of data is derived after applying some summary statistics, than even sampling the data works, and you can lose events in an event stream, while not changing the insight gained. Originally Kafka was designed to be a high throughput data bus for analytical pipeline where losing messages was ok. More recently they are experimenting with exactly once delivery.

vannevar 9y ago
Yeah, this was a major overstatement. There are lots of data pipelines where it's ok to lose some data. Consider a sensor that sends measurements hundreds of time a second to an app that operates on a 1-second timeframe. And UDP is used all the time on the internet, yet carries no delivery guarantee.

wonderwonder 9y ago

Would something like AWS SQS not scale for something like this? We currently push about 25k daily transactions over SQS, obviously no where near the scale of this, just wondering about what limitations we will bump into potentially.

psadauskas 9y ago
SQS is not "exactly once", so might not meet their requirements.
- idunno246 9y ago
  sqs has fifo queues which claim to be exactly once
  - lostcolony 9y ago
    SQS queues deduplicate over a 5 minute window. This is claiming a much larger window. Either way your listener(s) still has to have its own deduplication. Ensuring a message ends up on the queue only once, and ensuring it's processed exactly once, are two different problems that require separate handling (and, the former is what most out of the box systems claim to solve, while the latter is more important, and, frankly, completely negates the need of the former).
- twodave 9y ago
  Actually, it can be. You just pay more.
eropple 9y ago
(edit: incorrect, my bad, see thread)
- twodave 9y ago
  This is wrong. There is a type of SQS queue that indeed does exactly-once. It costs more, and not the default option, but it is there.

ju-st 9y ago

52 requests, 5.4 MB and 8.63 seconds to load a simple blog post. With a bonus XHR request every 5 seconds.

heipei 9y ago
That's not everything: This website contacted 56 IPs in 6 countries across 47 domains to perform 100 HTTP transactions. In total, 3 MB of data was transfered, which is 5 MB uncompressed. It took 4.103 seconds to load this page. 37 cookies were set, and 8 messages to the console were logged. https://urlscan.io/result/b2e27a08-1298-491a-863f-8cadc45e73eb https://urlscan.io/result/b2e27a08-1298-491a-863f-8cadc45e73...
joshuagvk 9y ago
And each one made only once!
numbsafari 9y ago
... and this is why friends don't let friends use Segment.

iampims 9y ago

If the OP doesn't mind expanding a little on this bit, I'd be grateful. > If the dedupe worker crashes for any reason or encounters an error from Kafka, when it re-starts it will first consult the “source of truth” for whether an event was published: the output topic. Does this mean that "on worker crash" the worker replays the entire output topic and compare it to the rocksdb dataset? Also, how do you handle scaling up or down the number of workers/partitions?

ryanworl 9y ago
I'm not the OP, but changing the number of Kafka partitions isn't a super graceful operation. You would be wise to add as many as you could reasonably need assuming one consumer thread per partition. But not too many because each one is at least two files on disk!
- tedmiston 9y ago
  I haven't used Kafka yet, but a Kafka partition is roughly the same as a Kinesis shard, right?
  - ryanworl 9y ago
    Yes, but you have to set them up upfront and there is no API to split and merge partitions. If you add more partitions, it doesn't automatically re-shard for you either. I don't think it should automatically re-shard as it would cause a ton of disk and network IO, but just something to be aware of.

robotresearcher 9y ago

> [I]t’s pretty much impossible to have messages only ever be delivered once. IIRC, it's provably impossible in a distributed system where processes might fail, i.e. all real systems.

mamon 9y ago

"Exactly once" model of message is theoretically impossible to do in distributed environment with nonzero possibility of failure. If you haven't received acknowledgement from the other side of communication in the specified amount of time you can only do one of two things: 1) do nothing, risking message loss 2) retransmit, risking duplication But of course that's only from messaging system point of view. Deduplication at receiver end can help reduce problem, but itself can fail (there is no foolproof way of implementing that pseudocode's "has_seen(message.id)" method)

malandrew 9y ago
An overview of FLP Impossibility http://the-paper-trail.org/blog/a-brief-tour-of-flp-impossibility/ http://the-paper-trail.org/blog/a-brief-tour-of-flp-impossib...
LgWoodenBadger 9y ago
I agree. I wish more messaging platforms would recognize this and stop trying to paper-over the failure mode (Kafka just came out with "Exactly Once," which I think is a variant of the 2-Phase-Commit protocol, which still does not solve the problem). My go-to for explaining to people is the Two Generals Problem https://en.wikipedia.org/wiki/Two_Generals%27_Problem https://en.wikipedia.org/wiki/Two_Generals%27_Problem
- LgWoodenBadger 9y ago
  After watching the video from the Kafka Summit talk https://www.confluent.io/kafka-summit-nyc17/introducing-exactly-once-semantics-in-apache-kafka/ https://www.confluent.io/kafka-summit-nyc17/introducing-exac... it's an essentially-useless guarantee for any sort of Kafka consumer that interacts with an external system. Exactly-Once is not something that can be provided-for or delegated-to an arbitrary, non-integrated system. For an "integrated" system, it requires idempotent behavior, in which case it's really At-Least-Once, so...
ecesena 9y ago
I think "exactly once" doesn't imply "non-reprocessing" (sorry for the double negative). Meaning, you want "exactly once" and you don't want duplicates, yes. But you allow for reprocessing, provided that you have a way for deduplicating. You want a guarantee that if the producer (at the top of your data processing pipeline) sends a message, then this message eventually corresponds to exactly 1 record in your final storage(s). One easy-to-understand-yet-simplistic example is: send a message to kafka, use topic+partition+offset as primary key, store in a RDBMS. This is widely accepted as "exactly once", but clearly you may have multiple attempts to save the message into the db, which will fail due to the primary key integrity constrain.

mooneater 9y ago

Awesome story. What I would like to hear more about, is the people side. The teams and personalities involved with coming up with this new system and the transition.

falcolas 9y ago

So, a combination of a best effort "at least once" messaging with deduplication near the receiving edge. Fairly standard, honestly. There is still a potential for problems in the message delivery to the endpoints (malformed messages, Kafka errors, messages not being consumed fast enough and lost), or duplication at that level (restart a listener on the Kafka stream with the wrong message ID) as well. This is based on my own pains with Kinesis and Lambda (which, I know, isn't Kafka). In my experience, better to just allow raw "at least once" messaging and perform idempotant actions based off the messages. It's not always possible (and harder when it is possible), but its tradeoffs mean you're less likely to lose messages.

caust1c 9y ago
This is generally better, but we're delivering these messages to integrations which don't necessarily take idempotent actions.

incan1275 9y ago

To be fair, they are upfront in the beginning about not being able to adhere to an exactly-once model. "In the past three months we’ve built an entirely new de-duplication system to get as close as possible to exactly-once delivery" What's annoying is that they do not get precise and formal about what they want out of their new model. Also, their numbers only speak to performance, not correctness. On the plus side, I think it's awesome to see bloom filters successfully used in production. That sort of thing is easy to implement, but not easy to get right for every use case.

qsymmachus 9y ago

It's funny, at my company we implemented deduplication almost exactly the same way for our push notification sender. The scale is smaller (about 10k rpm), but the basic idea is the same (store a message ID in a key-value store after each successful send). I like the idea of invalidating records by overall size, we hadn't thought of that. We just use a fixed 24-hour TTL.

bmsatierf 9y ago

In terms of connectivity, we deal with a similar problem here at CloudWalk to process payment transactions from POS terminals, where most of them rely on GPRS connections. Our network issues are nearly 6 times higher (~3.5%) due to GPRS, and we solved the duplication problem with an approach involving both client and server side. Clients would always ensure that all the information sent by the server was successfully received. If something goes wrong, instead of retrying (sending the payment again), the client sends just the transaction UUID to the server, and the server might either respond with: A. the corresponding response for the transaction or B. not found. In the scenario A, the POS terminal managed to properly send all the information to the server but failed to receive the response. In the scenario B, the POS terminal didn't even manage to properly send the information to the server, so the POS can safely retry.

goldenkey 9y ago
lòok maah no race cawnditions!!
v1shnu 9y ago
Why not just send the data too along with the UUID? It'll save another roundtrip in case of scenario B right? Or do you have data to prove that scenario B is a lot less likely to occur, making it sensible to save bandwidth by not re-transmitting the data?

linkmotif 9y ago

It's worth noting that the next major Kafka release (0.11, out soon) will include exactly once semantics! With basically no configuration and no code changes for the user. Perhaps even more noteworthy is this feature is built on top of a new transactions feature [0]. With this release, you'll be able to atomically write to multiple topics. [0] https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+E...

squiguy7 9y ago

I wonder how they partition by "messageID" they use to ensure that the de-duplication happens on the same worker. I would imagine that this affects their ability to add more brokers in the future. Perhaps they expect a 1:1 mapping of RocksDB, partition, and de-duplication worker.

squeaky-clean 9y ago
I'm also curious about this. I'm guessing it's some function on the id that maps it to a partition. What happens when you add more consumers? Is there a way to know which partitions include messageIds that would be put into another partition's ownership by the change and then move anything from one RocksDB instance to the other?
bytecodes 9y ago
Kafka does this as part of its design. A topic has a declared number of partitions (which can't really be changed on the fly, you choose a theoretical high number and hope it's enough), and an agreed upon hash algorithm chooses between those partitions (probably in Java, so hashCode is readily available for primitives as well as objects). Each partition is really like its own topic, so you lose in-order messaging for anything not included in your partition key.

gsmethells 9y ago

Why do I get the feeling this is repeating TCP features at the Message level? There must a protocol that can hide this exactly once need away. TCP doesn't create downloads, generally, that are bad and fail their checksum test, hence packets that make up the file are not duplicated.

jmaygarden 9y ago
How would you use TCP sockets to de-duplicate Kafka streams with a many-to-many communication pattern? Surely there is a valid scalability reason for why AWS IoT only provides "at least once" guarantees in their MQTT broker even when TCP is the underlying transport [1]. [1] http://docs.aws.amazon.com/iot/latest/developerguide/protocols.html#mqtt http://docs.aws.amazon.com/iot/latest/developerguide/protoco...
jkarneges 9y ago
Yes there is some duplication of TCP capability here. The problem with relying on TCP for reliability is that its state is in memory, associated with a particular peer IP address, and acknowledgements passed back to the sender only indicate that the receiver has the data in local memory, not that the data has been processed. A file download over TCP can fail, for example due to a network problem. Ensuring reliable delivery requires additional measures outside of TCP, such as retrying the download using a new connection. In practice, this means that TCP is primarily useful for providing flow control and offering a streaming interface (no worry about packet sizes). Less so as a complete solution for transmission reliability.

kortox 9y ago

With deduplication state on the worker nodes, how does scaling up, or provisioning new machines, or moving a partition between machines work?

PinguTS 9y ago

That reminds me of the safety-related protocols we use since years in embedded electronics like rail-road signaling, medical, and other areas.

spullara 9y ago

This isn't the solution I would architect. It is much easier to de-duplicate when processing your analytics workload later and you don't need to do so much work.

tejasmanohar 9y ago
That's not true for all the hundreds of real-time integrations that Segment sends data to. Many are write-only.
- spullara 9y ago
  Fair. For that case this solution makes more sense than the pure analytics case.

newobj 9y ago

I don't want to ever see the phrase "Exactly Once" without several asterisks behind it. It might be exactly once from an "overall" point of view, but the client effectively needs infinitely durable infinite memory to perform the "distributed transaction" of acting on the message and responding to the server. Imagine: - Server delivers message M - Client process event E entailed by message M - Client tries to ack (A) message on server, but "packet loss" - To make matters worse, let's say the client also immediately dies after this How do you handle this situation? The client must transactionally/simultaneously commit both E and A/intent-to-A. Since the server never received an acknowledgment of M, it will either redeliver the message, in which case some record of E must be kept to deduplicate on, or it will wait for client to resend A, or some mixture of both. Note: if you say "just make E idempotent", then you don't need exactly-once delivery in the first place... I suppose you could go back to some kind of lock-step processing of messages to avoid needing to record all (E,A) that are in flight, but that would obviously kill throughput of the message queue. Exactly Once can only ever be At Least Once with some out-of-the-box idempotency that may not be as cheap as the natural idempotency of your system. EDIT: Recommended reading: "Life Beyond Distributed Transactions", Pat Helland - http://queue.acm.org/detail.cfm?id=3025012 http://queue.acm.org/detail.cfm?id=3025012

Cieplak 9y ago
Basically the Two General's Problem, eh? https://en.wikipedia.org/wiki/Two_Generals%27_Problem https://en.wikipedia.org/wiki/Two_Generals%27_Problem
- newobj 9y ago
  Exactly.
  - jancsika 9y ago
    Question: what is the relationship (if any) between TCP and the Two Generals Problem?
    - 9y ago

travisjeffery 9y ago

Kafka 0.11 (recently released) has exactly once semantics and transactional messages built-in. - Talk from Kafka Summit: https://www.confluent.io/kafka-summit-nyc17/resource/#exactly-once-semantics_slide https://www.confluent.io/kafka-summit-nyc17/resource/#exactl... - Proposal: https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+E...

mshenfield 9y ago
My understanding (noob here) is that it allows producers to retry without fear of duplication. You still have to consider the system feeding the producer though. In Segment's example, clients might deliver their messages more than once to the API. Kafka's mechanism wouldn't detect duplicate messages sent to the producer, just that any given message a producer wants to append to Kafka won't be duplicated.

sethammons 9y ago

"Exactly Once" Over a window of time that changes depending on the amount of ingested events. Basically, they read from a kafka stream and have a deduplication layer in rocks db that produces to another kafka stream. They process about 2.2 billion events through it per day. While this will reduce duplicates and get closer to Exactly Once (helping reduce the two generals problem on incoming requests and potentially work inside their data center), they still have to face the same problem again when they push data out to their partners. Some packet loss, and they will be sending out duplicate to the partner. Not to downplay what they have done as we are doing a similar thing near our exit nodes to do our best to prevent duplicate events making it out of our system.

subhsandhu1 9y ago

Hawa Hawa Lyrics Mubarakan: http://www.desilyrics.in/2017/06/hawa-hawa-lyrics-mubarakan.html http://www.desilyrics.in/2017/06/hawa-hawa-lyrics-mubarakan....

ratherbefuddled 9y ago

"Almost Exactly Once" doesn't have quite the same ring to it, but it is actually accurate. We've already discovered better trade-offs haven't we?

majidazimi 9y ago

What is so exciting about this? There is still possibility of duplicates. You still have to put the engineering effort to deal with duplicates end-to-end. If the code is there to deal with duplicates end-to-end, then does it really matter to have 5 duplicates or 35? Or may be they just did it to add some useful cool-tech in to CV?

ngrilly 9y ago
> There is still possibility of duplicates. Where?

deleted 9y ago

[deleted]

vgt 9y ago

Qubit's strategy to do this via streaming, leveraging Google Cloud Dataflow: https://cloud.google.com/blog/big-data/2017/06/how-qubit-deduplicates-streaming-data-at-scale-with-google-cloud-platform https://cloud.google.com/blog/big-data/2017/06/how-qubit-ded...

redmalang 9y ago

Another possible approach: https://cloud.google.com/blog/big-data/2017/06/how-qubit-deduplicates-streaming-data-at-scale-with-google-cloud-platform https://cloud.google.com/blog/big-data/2017/06/how-qubit-ded...

siliconc0w 9y ago

Was thinking a 'reverse bloom filter' could be cool to possibly avoid the RocksDB for situations like this- turns out it already exists: https://github.com/jmhodges/opposite_of_a_bloom_filter https://github.com/jmhodges/opposite_of_a_bloom_filter I love it when that happens.

openasocket 9y ago
It should be noted it's impossible to have a 'reverse bloom filter' with the same properties as a regular bloom filter. That is, a constant predefined size gives you a particular false negative rate. The thing you linked to is really just a cache. It has to store an entire entry (not just set some bits based on the hash) and doesn't have a predictable false negative rate based on its size. For more info: https://cstheory.stackexchange.com/questions/6596/a-probabilistic-set-with-no-false-positives https://cstheory.stackexchange.com/questions/6596/a-probabil...

jkestelyn 9y ago

Relevant to this topic: Description of exactly-once implementation in Google Cloud Dataflow + what "exactly once" means in context of streaming: https://cloud.google.com/blog/big-data/2017/05/after-lambda-exactly-once-processing-in-google-cloud-dataflow-part-1 https://cloud.google.com/blog/big-data/2017/05/after-lambda-... (Google Cloud emp speaking)

skMed 9y ago

Having built something similar with RabbitMQ in a high-volume industry, there are a lot of benefits people in this thread seem to be glossing over and are instead debating semantics. Yes, this is not "exactly once" -- there really is no such thing in a distributed system. The best you can hope for is that your edge consumers are idempotent. There is a lot of value derived from de-duping near ingress of a heavy stream such as this. You're saving downstream consumers time (money) and potential headaches. You may be in an industry where duplicates can be handled by a legacy system, but it takes 5-10 minutes of manual checks and corrections by support staff. That was my exact use case and I can't count the number of times we were thankful our de-duping handled "most" cases.

luord 9y ago

This is interesting work. But I think I'll continue relying on at least once and idempotency. Exactly once is impossible anyway. > In Python (aka pseudo-pseudocode) This annoyed probably more than it should have.

openasocket 9y ago

So there's a lot of talk on here about the Two Generals Problem, so I thought I'd chime in with some misconceptions about how the Two Generals Problem relates to Exactly Once Messaging (EOM). WARNING: I'm going mostly on memory with this, I could be completely wrong. EOM is NOT strictly speaking equivalent to the Two Generals Problem, or Distributed Consensus, in an unreliable network. In distributed consensus, at some given point in time, A has to know X, A has to know B knows X, A has to know B knows A knows X, ... It has to do with the fact that the message broker is in some sense the arbitrator of truth, so the consumer(s) don't need full consensus. In an unreliable network, you can have EOM. http://ilpubs.stanford.edu:8090/483/1/2000-7.pdf http://ilpubs.stanford.edu:8090/483/1/2000-7.pdf gives some examples of how that works. HOWEVER, you can't have EOM when the consumers can fail. If a consumer fails there's no way, in general, to tell if the last message it was working on was completed. There are a couple of edge cases where you can still have EOM. For instance, a system where you have a message broker A, and a bunch of consumers that read messages x from that queue, compute f(x), and insert f(x) onto message broker B, where f(x) may be computed multiple times for the same x (i.e. if f is a pure function or you don't care about the side effects). This system can implement EOM in the presence of an unreliable network and consumer failures (I think it can handle one or both of the message brokers failing too, not 100% sure) in the sense that x will never be in broker A at the same time as f(x) is in broker B, f(x) will never be in broker B more than once for the same x, and any y in B had some x that was in A such that y = f(x).

philovivero 9y ago

tl;dr: Clickbait headline. Exactly-once delivery not even close to implemented. Typical de-duping, as you've seen and read about hundreds of times already, is what they did.

alexandercrohde 9y ago

Here's a radical solution. Instead of becoming a scala pro akka stream 200k engineer with a cluster of kafka nodes that costs your company over $100,000 of engineering time, technical debt, opportunity cost, and server costs, just put it all in bigtable, with deduping by id.... Enough of resume-driven-engineering, why does every need to reinvent the wheel?

azernik 9y ago
Yup. Databases, whether relational or not, have been designed to solve all these problems in a much more "bulletproof" way than your piddly [1] several-dozen-engineer team could ever manage, no matter how genius they are. [1] No disrespect meant - just a description of size. Source: running a piddly 2-person engineering team.

pfarnsworth 9y ago

Sounds very cool. A couple of questions I had: 1) What happens if they lose their rocksdb with all of the messageIds? 2) Is their kafka atleast-once delivery? How do they guarantee that kafka doesn't reject their write? Also, assuming they have set up their kafka for at least once delivery, doesn't that make the output topic susceptible to duplicates due to retries, etc? 3) >Instead of searching a central database for whether we’ve seen a key amongst hundreds of billions of messages, we’re able to narrow our search space by orders of magnitude simply by routing to the right partition. Is "orders of magnitude" really correct? Aren't you really just narrowing the search space by the number of partitions in kafka? I suppose if you have a hundred partitions, that would be 2 orders of magnitude, but it makes it sound like it's much more than that.

ngrilly 9y ago
> What happens if they lose their rocksdb with all of the messageIds? I'm wondering the same.