Streams: a new general purpose data structure in Redis
antirez.com
antirez.com
I've been thinking about this sort of thing for a while, wanting to maybe call it "slow code" (like "slow food"). This is how actual quality software that will stand the test of time gets designed and made, _slowly_, _carefully_, _intentionally_, with thought and discussion and feedback and reconsideration. And always based on understanding the domain and the existing software you are building upon. Not jumping from problem to a PR to a merge. (And _usually_ by one person, sometimes a couple/several working together, _rarely_ by committee).
I guess the phrase 'rush to judgement' is instructive. We're not just thinking, we're deciding of course.
If you can find out about a problem or issue or desire and very quickly come up with an architectural solution that will stand the test of time for years (at least without backwards-incompatibilities, which is what a redis requires), more power to you, but in my experience and observation most will not.
But I get it, you just don't agree with my basic suggestion, from your experience. That's fine. It's not really an argument about the speed of one's thinking. Nobody is "confused".
A good talk about It, from the creator of clojure:
No not really. It's called waterfall and we had it for decades.
And it was largely abandoned in favour of agile development because despite architects spending months designing something there was always something that was left out. Or the scope changed during those months. Or millions of other things that change whilst your hidden away from your customers/end users instead of delivering them new value every week.
For me personally the truth is somewhere between agile and waterfall. Some upfront design but not the slow code you refer to.
I also agree with the person who responded that there is a difference between building products, and building tools for building products -- building shared libraries/code/tools may require different approaches. One of the things I've noticed about shared code, is it's _much harder_ to fix rethought decisions or change paths. "Backwards incompatibility" doesn't even exist as a thing in your local app really, at least you can always at least theoretically global search and replace when you change APIs.
I don't think we've figured out any magic bullet process for developing software. It's definitely not the stereotype of "waterfall", but I don't think the stereotype of "agile" is it either -- at _least_ when it comes to shared library/tools. And I'm absolutely convinced that those shared tools which do well were almost always designed carefully and intentionally with understanding of the domain, not just slapped together with stimulus-response.
This is good. My version of "slow code" is taking note of requests from users and storing in my long term memory to stew over.
1. User A says "It would be nice to have feature X"
2. I ask for more info, what problem are they trying to solve
3. Repeat steps 1 and 2 until user's stop saying "nice to have" and start saying "must have". The stew is now ready and development can begin now that I have thought about a solution that can address everyone's problems...just in time before users get angry :)
The ID is composed of two parts: a millisecond time and a
sequence number. The number after the dot is the
sequence number, and is used in order to distinguish
entries added in the same millisecond.
Not sure if that's what they do, but it isn't very complex.
Something like: `XADD MAXSIZE ~ 2147484000 * foo bar` to cap the stream at 2GB + 1 node.
- UUIDs use a different epoch (15 Oct 1582 vs 1 Jan 1970) - UUIDs count 100 ns blocks instead of ms - UUIDs include a 6 byte "node id" - UUIDs allow only up to 15 bits of "sequence"
I think that last one is the biggest deal, since as currently specced redis allows 64 bits of sequence, which is obviously much bigger than 15. The options I see are either up the time resolution used by redis, encode some of redis's sequence bits into the UUID's time bits, or just live with it as a limitation--in practice 2^15 is a lot of messages to get in a single millisecond (though in cases of clocks jumping back might not be too much).
You'd also need to come up with some thing for the node id, perhaps the first 6 bytes of a cluster node ID or similar.
Since 64 bits is overkill for milliseconds (45 bits covers the next 1000 years or so) I was thinking you could put 2 bytes of the node id in the high order bytes there (perhaps could call this the "clock id"?) and the remaining 4 bytes of the node id could go in the high order bytes of the sequence, which would still leave 32 bits for actual sequence values (but we should only use 26 or so). This means we'd get a translation roughly as follows (numbering bytes and bits from high to low significance):
Redis Version 1 UUID
Timestamp
Byte 0-1 "clock id" Bytes 4&5 of node id
Byte 2-7 millis since 1 Jan 1970 * 10000 => ~45 high order timestamp bits
Sequence
Byte 0-3 "node id" Bytes 0-3 of node id
Byte 4-7
6 bits wasted space ignored
26 bits actual sequence value
13 high order bits => clock sequence
13 low order bits => low order timestamp bits
On a side note it seems that the clock sequence in the UUID is intended to be reset to a random value at start up and every time a clock jump is detected rather than just incremented. Redis could do something similar by incrementing some of the 13 high-order bits of the sequence every time a clock jump is detected (and/or if the 13 low-order bits overflow)
[0]: https://github.com/antirez/redis/commit/1189d90d749c84e98424...
Example:
c - h72gsb32 - 0000 - udoc - l363eofy
The groups, in order, are:
1. 'c' - identifies this as a cuid, and allows you to use it in html entity ids. The fixed value helps keep the ids sequential.
2. Timestamp
3. Counter - a single process might generate the same random string. The weaker the pseudo-random source, the higher the probability. That problem gets worse as processors get faster. The counter will roll over if the value gets too big.
4. Client fingerprint. For example, the first two chars are extracted from the process.pid. The next two chars are extracted from the hostname.
5. Pseudo random (Math.random())
I think antirez is saying that, by serializing the id in this way, you can also get timeseries at the ms precision for free.
Edit: nvm, antirez just replied :)
As for treating them as decimals inadvertently, well, hopefully client libraries will expose IDs as pairs of integers, not as strings. If users convert them into strings and then back into meaningless pseudo-decimals, well, great, we'll have an entertaining post about someone's outage to read.
For versions with only two components, I would argue that the dot can be confusing already.
Why use a separator that has the potential of confusion when there are several other choices with less potential?
Once the specification is stated clearly as it is right now it is not a problem.
Another symbol could be used (for instance #) but I really don't see a need for that.
A good reason not to change it from a period to a comma, not really relevant to whether to change it from a period to something else.
> I agree with the other commenter that this is no different than using periods in IPv4 addresses.
IP addresses have no useful concept of "before" or "after", whereas these do.
XADD mystream * sensor-id 1234 temperature 10.5
to let the server choose an id.
It could also be interesting to have
XADD mystream 1506872463535.* sensor-id 1234 temperature 10.5
to be able to add an element in a specific msec bucket
I wonder if this opens such projects up to disruption, in the original sense of the word. I've seen the same teams forgoing apache for nginx, then forgo nginx for haproxy when it matches their needs. With additional layers of complexity, there may be accompanying opportunities for "simple but good" projects to gain traction.
I don't think it's very avoidable. Tech is genuinely getting incrementally better, but it's usually in a sawtooth pattern.
My favorite aspect of this cycle is when some features in the complex software become seen as so useful as to be required and standard, so when the new simpler version is created they have to figure out a novel way of providing that useful functionality in a simple and elegant way. And they do it, sometimes knowing they have made a significant advance and sometimes without knowing.
Sounds too good to be true... but I'd love to be proven wrong :)
A new data structure + associated commands being supported in Redis is like, say... a new filesystem being supported in the Linux kernel. It's a few files that you could just avoid compiling in if you didn't want them, and which add code-paths that are never run unless you intentionally use that specific new thing.
> A new data structure + associated commands being supported in Redis is like, say... a new filesystem being supported in the Linux kernel. It's a few files that you could just avoid compiling in if you didn't want them, and which add code-paths that are never run unless you intentionally use that specific new thing.
Isn't it exactly how features are implemented in Apache? Most people think that Apache is too bloated, but the same people also leave a lot of those modules enabled without trying to spend time to understand if they actually need them.
Redis modules, like filesystem drivers, are comparatively simple: they register a set of commands that they respond to, and each such command is handled exclusively by that module when received. Less like Apache plugins, more like scripts in a cgi-bin. You don't need to know what else is in the cgi-bin besides your own script, because everything there is entirely independent.
Another data point, streams are if not completely, at least 70% done as we talk, and yet `git log --stat unstable..streams` reports:
12 files changed, 1323 insertions(+), 10 deletions(-)
We'll end at 2000 lines of code I guess more or less. Just to say, Redis is at a complexity scale very far from many other things we are used to these days.
Obviously the general thinking is that event sourced time series data allows you to treat every other data source as derived state from the log data. If it gets written to the log, it's safe and that's the only real data that cannot be considered a redundant read layer. A common structure might look like:
1.) Event Sourced/Time Series Layer: Kafka/Kinesis>S3 takes in and saves log data
2.) Operational State Layer: RDBMS Constraints / Application Logic determine how operational state is derived from log data
3.) Indexing Layer: Query optimizations occur with redundant read layers in what is essentially all just various forms of indexing. This can be RDBMS index, ElasticSearch, MapReduce, Redis, etc.
Redis' place has historically been at #3, for many reasons. Whether an application has an event-sourced layer in which their operational state is derived from log data or is actually considered the primary source is something that is hugely variable. I would say that most applications do not make a distinction between #1 and #2, and just write state directly to an RDBMS. But while the operational state may or may not be considered redundant, depending on the application, the indexing layer is almost guaranteed to be a redundant layer. The redundancy of the index layer means that Redis operating purely in-memory allows the whole thing to be blown away with no consequence. To move it two steps down the data model to the defacto source of truth is a monumental shift in responsibility. Redis as an in memory caching layer has only ever had me have a cursory awareness of its capabilities in terms of saving to disk, but I would think that a fundamentally different use case like this will have me taking a serious look at where that functionality is at today.
All of this being said, there are plenty of use cases with kafka/kinesis which are done today which actually don't even save the log data at all, and just use them as an intermediary buffer to have multiple consumers on an event stream. There's also nothing stopping us from just having one of the consumers of this stream to be saving it to S3/Disk ourselves.
If not, either people's messages don't matter /that/ much (which is fine, just not great for most of my usecases at the moment) or everyone's in for another round of "oh shit, where did the data go?"
Edit: Just in case we end up in CP vs AP datastore wars, please go read https://martin.kleppmann.com/2015/05/11/please-stop-calling-...
At-least-once delivery requires neither CAP consistency (linearisability) nor CAP availability (any non-failed node must return a response in a non-infinite time), but is a very useful property!
With vanilla Redis master/slave replication, I believe the best way to avoid data loss is to set replication to be synchronous (it's async by default) so that slaves are always guaranteed to be in sync with the master, in case you need to promote (using Sentinel) a slave to master.
1. Use the default asynchronous replication, and live with the fact (if the use case permits this) that on failover, the message did not yet received the slave that will be promoted.
2. Use WAIT to force synchronous replication to N slaves. This will not still make Redis ensure you in mathematical terms that the failover will pick a slave that received the message, under complex partitions, but narrow the real world failure models leading to losing data to more "unlikely" cases. Yet you have just best effort consistency but with better real-world outcomes.
So Redis streams will be good choice if one of the above is acceptable.
Not currently useful to me, but I'm sure this hits the sweet spot for a bunch of people.
It sounds like anything can be lost in redis during normal HA operations even with WAIT pushing to a majority of slaves. Is that right?
EDIT: And as antirez said above, you can WAIT to force synchronization to all nodes, so you would be pretty likely to fail over onto a node that has n-1 messages if it didn't sync in time. That still isn't guaranteed however.
1. Consider adding an example for a stateful event stream processor client that saves the last read stream offset in redis, together with its current state and continues reading from that offset as an atomic operation. For example, a client that sums a stream of numbers, in order to have effectively once semantics would need to persist to redis the sum and offset together.
2. Consider adding a stream read deduplication example to mitigate clients that reinserted the same event twice. It is not clear how the client should behave if it didn't get an ack and it resents an event. What is the correct resending semantics so the reader would effectively dedup? What is the right data structure used to dedup message ids without consuming too much memory, etc...?
As a more generic solution, there's also pushpin (http://pushpin.org/), which is the backend of fanout (https://fanout.io/), so that may also be a nice addition to your stack if you want a more direct redis->clients link
Also, all the tooling around it is quite mature - there are great monitoring and management tools for probing at the internals which helped when we were dealing with more exotic kafka surgery.
People say that setting ZK cluster is a huge issue, yet they don't see a problem spinning etcd, or sentry nodes in case of redis.
When I learned about ZK I was skeptic, didn't like that it was written in Java, but ZK proved to be extremely robust.
2 - Kafka clients are not available (or are poorly available) for every programming language. Redis has a simpler protocol, so it has more/and better clients available and even if you use an exotic language, it is easy to write a client to it (well... easier than Kafka)
3 - Kafka AFAIK does not have any internal cache implementation, so every read is served from disk (+ page cache). This means that Redis Streams will (probably) perform much better for use cases when the consumers need to fetch data from old offsets.
edit: added reason number 3.
Most Kafka client implementations are open-source projects of their own, this is also true for most redis clients implementations, but again: Kafka protocol is much more complicated than Redis.
I haven't used Kafka with other languages besides Java or Scala, so I can't really say how mature are the other clients.
But my point about how easy is to implement a client for Redis if needed is still valid. =)
I really can not see how Redis is going to perform "much better" reading from disk once the entries are no longer in RAM. At that point both Kafka and Redis have to read from disk, and you either have the IOPS to serve all the lagging consumers or you don't. Maybe you have enough of them to service 1 or 2 concurrent reads, maybe 10-12. But for the same messages counts, sizes and concurrent consumers, your workload will become IOPS bound rather fast.
Note: "much better" to me implies 10x+ better, not "my C library read() is 2.3% better than your Java".
<meta name="viewport" content="width=device-width, initial-scale=1">
#content { max-width: 800px; } // replaces width: 800px
A good tool should be able to be used at any scale.
I don't necessarily agree that a tool should be used at any scale even if it's technically possible to. Cost (multiple dimensions, including money and engineering effort) factors in.
Kafka also has the JVM, which requires a lot of love to scale in my experience. I do not want my programmers messing around with GC options when writing to what should (to them) be exposed just like a regular file handle (except distributed across many systems). I strongly prefer to avoid Java applications at all costs - in my experience it takes years and years and years for Java based infrastructure to become relatively stable & reliable (see ElasticSearch 5.0, or ask anyone who has been oncall for a Tomcat based application). This is almost certainly personal bias, but it's my bias regardless.
Redis also has a _massive_ number of tooling / monitoring / ecosystem advantages, including hosted options, and can run on a single instance without configuration changes from the developers perspective.
I also have personal reasons to prefer Salvatore's work over the work of Confluent.
So, a Zookeeper cluster can't survive accidentally injecting just the right malicious data that will make it keel over. I'm sorry, how do you accidentally achieve that? Do you also accidentally configure your Redis Sentinel to replicate from /dev/null?
As a matter of fact, this announcement comes at a very inopportune time for me. antirez had the epiphany of reading on IRC about replicated logs instead of looking at the opening paragraphs of the Kafka documentation, and all the Redis evangelists at my job will now try to shoe-horn the wrong usecase back into Redis because Redis!1cos(0)!. Sigh.
How does that happen? I mean a new, empty ZK server with never data than the rest of the cluster?
Also, please note that ZK is not meant to be a database, but a coordination service, it's guarantee is to have all nodes being always in consistent state and neither of its nodes allow to make any changes if there's no quorum. So if a new node somehow has more recent data with higher serial number it's expected that remaining nodes will sync to that.
I don't hate Kafka, I just don't like ZK and find redis has better tooling and a better track record at my shops :)
It's hard to accidentally add node to a cluster. A person who can "accidentally" add a ZK node has enough permission to do a lot of more devastating things accidentally.
This is all in service to my point about simplicity and safety.
This is almost the exact opposite of how I determine what tools to use. If it's written in C or Java, I'm usually pretty confident that it is engineered by a team of experienced developers. Both because the languages are technically more difficult to use, and not as cool.
In contrast, if a tool is written with javascript (Node) or Ruby, and often times Python, I'm very hesitant.
In fact, this whole topic of streams has me wondering just how many developers out there are setting up clusters of Kafka or Redis or whatever is new and hip, when they could have saved themselves a huge amount of pain by using tried and true tools like JMS or ZeroMQ.
Most companies do NOT have a need for scaling like Netflix or LinkedIn, and I'm beginning to wonder if Kafka, Redis, etc are this year's version of NoSQL and MongoDB hype.
I love redis, so I certainly don't have issue with code written in C, heh. Just code written in C by junior developers :P
I can already see lazy users just repeatedly reading $, and then dropping messages when they arrive faster than they read them.
Might it be safer to instead have command to ask what the latest ID in the stream is? You'd start off by using that to work out where the streams are, then construct an XREAD command to read from there. To construct your next XREAD, it should be easier to update the IDs from the ones you just read, rather than fetching the latest IDs again. Maybe.
https://engineering.linkedin.com/distributed-systems/log-wha...
The Kreps article is quite excellent and well worth the read.
Love it. No need for an expensive SCAN command.
Have you considered adding a semantic layer inside streams that allows each client to consume a substream? In effect the stream becomes multiplexed substreams.
If substreams makes the design too complex... have you considered server side stream 403 semantics? When a stream is manually deprecated it enters an immutable state and provides a redirect response with a link to another stream. This would allow multiplexing and demultiplexing streams without changing the client implementations too much.
For completeness I would state the obvious when fifo grouping is needed: 1. Scaling stateful event processing by splitting streams and adding more clients (CPU limit) 2. Scaling cross region replication by splitting streams and adding more tcp connections (network limit) 3. Handling more throughput by splitting a stream into two redis nodes (disk I/O limit)
Run multiple Redis instances and use a simple hash based on whatever key you want to route messages and get the throughput you need. It's probably never going to be part of the core Redis logic but should be possible as a module to do the routing when used in a Redis cluster.
Doing it without server support is tricky.
I'm kind of thrown by the idea of putting the timestamp / stream-id in the XADD command, I would have thought the server would assign that, since one of the strengths of redis's single threaded nature is consistency: what's in redis is the truth. If you allow clients to specify timestamp, what happens when ntpd isn't running on some? I probably misread or misunderstood it.
Could you allow specifying `$` as the timestamp to tell the server you want it to use whatever it thinks the current time is as the timestamp / stream-id?
127.0.0.1:6379> xadd stream * a 1 b 2
1506977609865.0
"xadd","stream","1506977609865.0","a","1","b","2"
127.0.0.1:6379> xadd stream 10.0 a 1 b 2
(error) ERR The ID specified in XADD is smaller than the target stream top item
The reason why it was chosen to use milliseconds instead of nanoseconds is because, for most applications to query for sub-millisecond ranges is likely not useful, so to see even larger numbers in the ID maybe is just unpleasant if not useful, however we are still in time to change this if there are good motivations. But being the time the one produced by the local host, after a failover the IDs are generated by another host. Milliseconds can still more or less match with good time synchronization, but nanoseconds? So it's like if this additional precision will be just used to store non-valid info.
I was reading on mobile earlier and maybe missed this point, excellent. I also didn't realize that clients would use (star) to specify the ID and that the receiving server turns that into an actual ID before replicating it / AOFing it.
> But being the time the one produced by the local host, after a failover the IDs are generated by another host. Milliseconds can still more or less match with good time synchronization, but nanoseconds? So it's like if this additional precision will be just used to store non-valid info.
I think most failovers necessarily take longer than a millisecond so any resolution smaller than millis would _probably_ be okay, but yeah this is not a compelling reason to switch to micros/nanos. My suggestion to switch to micros/nanos was more to try to reduce the number of collisions requiring the server to de-dup / assign sequential sub-epoch numbers to events arriving during the same server tick. I guess that's not a big issue though.
Thanks for the reply, Salvatore. Redis is one of my favorite codebases and projects.
http://code.flickr.net/2012/12/12/highly-available-real-time...
I currently use multiple Sorted Sets (one set every 5 minutes) and Union 30-60 worth to produce a "rolling window" of uniques.
I can see an alternative where I just request a unique count of elements within an XRANGE.
Use case: A log of network latencies, where a user might currently `XREAD` with a timestamp 10 minutes in the past, would be able to save on memory usage by expiring log entries > 10 minutes, and then being able to `XREAD STREAMS strm 0` and let Redis (and therefore the infrastructure, not my code, manage data retention).
Also, how does this work re: evictions? Say a node is at max memory, will entire _streams_ be evicted, or (I hope) the oldest messages in the LRUed or LFUed queues.
many clients will even not need that timestamp anyways.
I would humbly suggest that "^" would be a suitable symbol for the first message in a stream. ^ and $ are used in regex (and vim) in a similar way.
That way you could write "XREAD BLOCK 5000 STREAMS newstream ^" and get all the messages in a stream from the beginning, and then block until a new message comes in all with a single command. You would still be able to add a count if needed, to prevent client flooding.