RailsConf 2014 - Supercharge Your Workers with Storm by Carl Lerche

CARL LERCHE: My name is Carl Lerche and

I'm gonna be talking about Apache Storm.

I am shooting for forty-minutes.

It's going to be a long talk.

I'm gonna try to cram everything in.

I already got a late start.

There probably won't be time for questions.

I'm also probably going to gloss over

a few things that, like, hand-wave-y things,

so if something comes up, you have a question,

just send me a Tweet directly

and I will respond to it after the talk.

Or just come and talk.

So I work for Tilda, and our product is

Skylight, and we're building a smart profiler
for your

Rails app. And I've actually built the entire
backend

using Storm, so another thing, if you want
to

talk about how that works, how we've, I've
ended

up using Storm and the, in the real world,

I guess. We, I'll be happy to talk about

it.

So what is Storm? Let's start with how it

describes itself on the web site. It calls
itself

a distributed real time computation system,
and that sounds

really, really fancy. The first time I thought
about

using it, I was like, oh, let me go

look and see what Storm's all about. I saw

this, was like, whoa. Step back. I'm not going

to. This, this sounds like, this is too serious

business for me. I need something simpler.

It took me like maybe like another six months

of really, OK, I'm gonna take time, I'm gonna

learn this, and as I got to know it,

it really came to me. Oh, this can be

used for many, many other use cases. I'm not

going to. I might, at the end, if there's

time, talk about some of them, but I'm going

to try to get into the nitty gritty sooner

than later.

So, but for now, I'm just gonna say, Storm

is a really powerful worker system. So some
things

about it. It is distributed. It's very, very,
very

distributed, which is a good and a bad thing.

One, the good thing is you can, like, spread

your load across many servers. The bad thing
is,

your operate, like your ops system is gonna
look

something like this. Which can turn out to
be

somewhat of an operational headache. So, like,
you'll end

up having to run a Zookeeper cluster. You'll
end

up having to run, like, the Storm Nimbus process

somewhere, and then there's gonna be a number
of

other Storm coordination processes. There's
gonna be, like, every

server's gonna have the worker process.

And then, like, while you're at it, you've
gone

this far, let's throw in a distributed database
and

some distributed queues in there too. But
the main

point is there's going to be an operational
overhead.

So don't just do it for fun. There needs

to be a good reason. If you can get

the job done on a single server, go for

that.

It's fault-tolerant, and what I mean by that
is,

it is able to recover and continue making
progress

in the event of a failure. And I know

a lot of existing systems claim to, claim
to

do this, but the reality is, pandering faults
is

really, really hard. So it doesn't make it,
like,

seamless. Storm doesn't make it seamless.
The reality is

it can't be seamless. But I think, and I'm

going to talk about it later, I think the

way it does it is about as easy as

it can get. I hope.

It's really fast. Very low overhead. Some
completely useless

benchmark numbers. It's been clocked at, like,
processing over

a million messages per second on a single
server.

Completely useless number but it sounds good.

Supposedly it's language agnostic. And by
that I mean,

supposedly you can use absolutely any language
at all

to build your data processing pipeline. There
are examples,

on the website, where they use bash to, I

don't know why, it's probably a novelty. If
you're

distributed bash sounds good. But I guess
that means

you can probably use MRI. However, I've personally
only

used it on the JVM. It's a JVM-based project.

It's written in Java and Clojure. It uses
many

Java libraries. I would just recommend, if
you're gonna

try it, try it out on the JVM.

And the good news is, we have JRuby, which

has done an extremely good job at integrating
with

Java. So you can use these Java libraries,
and

you're just writing Ruby. Some of the things
I

want to handwave also is exactly how to get,

like, all the details of getting it running
on

JRuby. But there is a GitHub project called
redstorm.

You could probably just Google GitHub Redstorm.
I've realized

I didn't actually include any links. But,
again, I

can find them later.

So it does a JRuby binding to storm. It

also goes a bit further. It provides, like,
Ruby

DSLs for everything related to storm. So you
can

define all your data transformations in terms
of Ruby

DSLs.

So the way I'm gonna kind of go over

storm is I'm going to build up a straw

man. That straw man is going to be, how

would one implement Twitter trending topics.
You know, it's,

I think I, we all hopefully know about what

a trending topic is. That's kind of why I

picked it. It's. But. Essentially, everybody
Tweets and they

can include these hash tags. Hash tag RailsConf.
Hash

tag whatever. And as the rates of single hash

tags increase, like the highest, the most,
the hash

tags that occur at the highest rates end up

getting bubbled up, and Twitter says, ah,
these are

the topics that are trending so that everybody
can

see what's going on.

So, like, around here, you might see, RailsConf
is

trending. Another good reason, a reason why
I picked

it is it's very time-sensitive, and times
into, ties

into the real time aspect of Storm.

All right. So to implement this, the way I'm

going to calculate the rate, like how often
a

Tweet is happening is by using an exponentially
weighted

moving average. Fancy. Basically what it means
is like,

as we're moving up, going, like, processing
the Tweets,

what I'm gonna do is I'm gonna count the

number of occurrences of each hash tag, for
like,

every five seconds. The interval doesn't mat,
really matter.

Then we're gonna average them up.

And, but instead of doing a normal average,
which

is sum up all the values divided by the

number of values, we're going to do a weighted

average, which is going to cause the older,
like

the older occurrences, to drop in weights
exponentially. This

is how Linux does their one minute, five minute,

and fifteen minute load averages.

It's a pretty nice way of smoothing out the

curves, and you can tune how you want the

curves to end up getting smoothed. So first
is

kind of build up some context. Let's kind
of

look at how one might implement this with
Sidekiq

or, like, Resque or whatever. Any other, like,
kind

of simpler system that just pops off of a

queue with a worker.

This is what it might look like. You've got

your Rails app. Pushes data into Redis and
then

each, like, the worker says, OK, I'm gonna
pop

a message off of Redis. I'm going to read

straight from the database. I'm gonna process
the message.

I'm gonna write straight back out from the
database.

And then the Rails app can just query the

result by reading it straight from the database.
Here's

a, it might run, I did not. I, is

that bright enough? Hopefully it is. The comment
basically

says, yes, this is very naive. I know it.

But that's not the point. Don't say, oh you

could do this one million times faster. That's
not

the point.

Hopefully it's readable.

So tags. So the basic premise is you have

an entry method, which takes in a Tweet as

an argument, and it, first step is to extract

the hash tags from the body. So I get,

like, an array of tags, and I loop over

each tag and I read from the database, is

there already an existing record? If so I'm
going

to update the moving average for it and then

save it back.

Oh. Yeah. Put highlights. So this is how one

might. It doesn't really matter. It was kind
of

like for informational purposes. But this
is how one

might implement the moving average like function.
So the,

the most important part that I want to call,

like, draw attention to is that in the update

EWMA, the very first thing we do is catch,

catch p. And I didn't pass now. But in

theory it should pass now down to the catch

up and it will keep ticking to update the

rate for every bucket that we decided we're
smoothing

by. So, like, every five seconds it's going
to

reupdate the int, the rate so that it will

downweight previous values and go on from
there.

So this is going to expose a problem with

our first version of the worker, in that if

we do not receive any Tweets with a given

hash tag, what, how do we handle that? We

need to still update the record in order to

constantly update the rates so that, ah, we're
not

resaving any data. We have to start re, lowering

the weight of that hashtag.

So, just, we'll have to end up implementing
something

like this. Which is going to be scheduled
via

Chrone or just regular, some regularly scheduled
task that

probably has to run every bucket interval.
And first

step is just delete all the existing tags
that

are way too old just to clear the database.

And then load up all the tags and update

the rates for them. Make sure they're caught
up.

So this is going to, one thing to point

out is, OK, so we now have to have

a task that is going to read literally every

record from the database every tick. And there
probably

is a better way to do this. But that's

not the point.

So is it web scale, yet? Well, let's look.

So far, not yet. But what we can do

is we can start scaling out our workers. We're

just gonna like, ah, let's spawn up three
workers.

Now we're ready to handle fifty-thousand Tweets
per second.

Well, maybe not, but we have more tricks up

our sleeve. Like, we know, like I mentioned,
we

know that the database is going to be a

pretty big bottleneck, cause every single
Tweet we get

in, it's going to have to read state, process

it, and write it back.

What if we could optimize the, and, remove
the

read state and, by caching a memory. Super
simple

implementation, which more is represented
as like, oh, unless

like, we basically build an in memory cache
of

hash tags in the process, such that every
time

we get a Tweet, it comes in and we

first check our cache before trying to, making
a

new one. And we update that and save it

back to the database.

So let's run through this and see what's happening.

Here's the first Tweet that comes, that comes
through.

We include, queue up, it comes in through
the

queue. Fancy animations, yeah. And it goes
to the

first worker. The worker processes it. It's
like, OK,

my cache is empty. I'm gonna check in the

database. Nothing for RailsConf in the database.
So I'm

just gonna process that memory the count one.
And

we're gonna write it to the database. K.

Next Tweet comes in. Ah, what's for lunch
at

RailsConf? Well we know now, but it's going
to

be, like, queued, like, again, queued up in
here,

and I'm sure you know where this is going.

Goes to the second worker. Second worker reads
the

counts and it works, increments count to two
and

it goes back and writes the database.

And here's the third Tweet. And we're gonna
get

to it. This one's going to go back to

the first worker, it's like, oh, I have everything

cached. I don't need to read from the database.

No probably. We'll count to two and yes. Yes.

This. I did this way too late. I was

like, ah, I'm just gonna do fancy animations.
All

right, so, caching in this kind of system
is

not obvious. We have this problem of, like,
we

have, we can still make it work. There's many

things we could do. Like, we could push, like,

we could push for cache external. We could
push

for mem cache, but now we still have an

external process that is required for coordinating
the work.

So, the main thing is, even though we have

many worker jobs, there is still a high amount

of coordination that's required to get this
to work.

K.

Enter Storm. And, as you can guess, probably
this

is gonna, like, all these problems are gonna
magically

go away when we use storm. I wish. But

it'll, there's something we can do. Let's
start with

some abstractions. The core abstractions in
Storm are the

streams and the tuples. A stream is, essentially,
a

series of tubes through which data flows.
And, but

they call them streams. I would, tubes would
have

been better.

And tuples are just, it's just like a list

of values. And these values can be anything.
They're

really just the messages. So you can put in

tuples, you can put your strings, you can
put

integers, you can put any object you want
as

long as you can serialize it.

And you, you're allowed to specify custom
serialization. So

as long as you can serialize it to JSON,

serialize it to any serialization format,
you can put

that object in the tuple.

The rest of Storm is going to be a

series, a bunch of primitives to take these
streams

and tuples and to transform the data from,
like,

one representation to the other.

Next, spouts and states. So the spouts are
the

source of the streams. So this is how you

get data into storm. This is the starting
point

of the streams. So anything that you want
to

read from the outside world is going to end

up being represented as a spout. And this
can

be reading from redis, reading from SQS, reading
from

anything. But it can also be, you could implement

a spout that reads directly from the Twitter
API.

You could implement a spout that reads from
the

database, that makes HTTP requests, that gets
the time

of the day, even. And, yes, you will want

to actually implement a spout that gets current
time

of the day. And I'll talk about that later.

States are the opposite. State is how you
get

the result of the transformations outside
of Storm. So

exactly the opposite. If you want to write
outside

of storm, like you write to the database,
write,

make HTTP post requests, send email, push
to external

queues, anything like that. It's going to
be implemented

as a state. So, so far this is what

we know about Storm.

We have Spout. It starts a stream. Data is

going to flow through it and it's going to

end up at the state. So far nothing is

super interesting.

Transform. The rest is really about how do
we

transform the data as it flows through. And
transforms

are gonna be purely functional operations
on the data.

So they are going to be given the, given,

given inputs, the same, it's, they're going
to output

the same values. Anyway.

So let's add a couple transforms here. So
we

have a spout. It's going to send some data

and we can, like, filter if we need to.

So we'll filter some data out and we'll aggregate

it. And the data will flow through and end

up at the state and then get persisted.

And then more transforms. Basically from here,
what you

end up doing is modeling your data in terms

of dataflow to get to the end point. And

there's really, you can, there's no limitation
in terms

of, I mean, granted you don't add a million

spouts because that will probably be crazy
and you'll

be, not be able to understand the code. But

besides that, you can add as many spouts as

you want. You can add as many state end

points. You can, like, split the streams.
You can

join them. And do whatever real transformation
that you

want. And Storm is going to take care of

figuring out how to take this definition,
like, the

state of the definition and distribute it
across your

cluster.

And this entire set, basically, wrap of starting
at

stream, starting at spouts and ending at states
is

called the topology. And in a bit, I'm gonna

show how to define the topology.

So, OK. Let's break it down just a little

bit. So the spout is where it starts, and

then it's going to emit tuples to the filter.

Generally, as you're gonna get the spout,
like, it's

gonna be a pretty standard thing. Like, for
example,

if you are pulling, if you are pulling data

from Redis and want, you will use a Redis

spout that already exists. And it will literally
just

be in your definition, just use Redis spouts.
And

filters, as well, any, any transforms can
usually be

abstracted up to higher level concepts and
shared.

So, for example, this is how you might implement

the my filter, the my filter function or transform

for Storm. So the base function is provided
as

part of the Storm API. And the requirement
is

that you define a single method that takes
the

input as a tuple and it will take the

output, which represents the stream. So you
will get

in a tuple, you're gonna process it, and you

will then emit zero or more output tuples
onto

the output stream.

So, the first step in the filter is just

get the, this is, get the message out of

the tuple, and this is somewhat of a Java-ish

API, but it's possible to shim. So, get value

by field. You, in this case, we're just creating

the tuple more as a hash map than anything

else. So we get the message. Is the message

awesome? If so, we just output it on the

output stream. And that is all that it takes

to implement a very simple transform.

K. Here's another example. And this is not
something

to run in production but is super helpful
for

debugging. All right, the next step is define
the

topology. Like, define how all the different,
like, transforms

and spouts and everything hook up together.

And the important part's on the screen. So
the

define_topology method doesn't matter, it's
whatever. The main points

you will start at, you will get a topology

object, and on that you'll use a series of

APIs define how things flow together. So you
start

by defining the stream and you just name it

as, are my-spout, and in this case, I'm saying

MyQueueSpout, but it would be like My, like
RedisSpout

dot new specify the server, the, like, wherever
the

server is and the topic you want to read

from. Or, if you're using SQS you would pass

in your credentials and the topic you're reading
from,

et cetera.

Next step, usually the spouts will just output
tuples

that are raw bytes. So it will just get

the raw bytes from Redis and output it, so

the next step is to deserialize the raw bytes,

so we are getting a message that, we are

saying ah, we are expecting in, as an input,

a tuple that contains the field bytes. We
are

gonna pass it to the queue message deserializer.
And

the output is gonna be a tuple that contains

message.

I included the input, an example implementation
under. And

then we're gonna chain that, and we're gonna
say,

like, for each tuple that we get, we're gonna

expect, well, we're gonna expect a tuple that
contains

the field message. We're gonna pass it to
my

filter, which will filter it based off of
our

predicates, and we'll output another tuple
that contains a

message. And finally we'll pass it the logger.

Next step, run it. And the easiest way to

get this running, just for exper, like, playing
around,

is just running it locally. And redstorm does
this,

but if you want to just, like, the actual

AP, like, amount of code to get it started

is ba- is just this. You initialize a new

cluster object. You define new TridentTopology.
I'm going to

just kind of like, wave my hands over what

Trident is. You can read the Wiki.

And doesn't really matter for the talk. You
can

initialize a config and then there's our define_topology
method.

You pass in topology and then you submit it

to the cluster, which is just your locally
running

machine. And the entire thing is running.
Winning.

But only a little bit winning, because we
have

not, we're not doing anything with the data
yet.

So the next step is going to be to

persist the results. And, again, there are
higher-level things

that, there are like, there are already existing
libraries

where it's like, oh, you take in these tuples

and just throw them, throw them entirely in
Redis

that take no code. But I am going to

jump directly down into how you would implement
a

state from scratch.

So what it takes is, you just make a

class, inherit from state. This is provided
from, as

part of the Storm API as well. The only

methods you have to define are begin commit
and

commit, and I'm gonna cover those later. Besides
that,

you just provide whatever you want. And as
an

API to interactive to state. Because next
component is

going to be the state updater.

The state updater takes in the, requires you
to

define the method update_state, which is going
to pass

in the instance of your state and is gonna

pass in the input tuple and an optional, like,

it also gives you an output string in case

you want your state to emit more tuples. But,

in our case, we're gonna, all we're gonna
do

is ex- get the message out of the tuple

and call persist awesomely, and that's it
for now.

Except, cause it's Java, it's also a factory.
But,

let's not mention that. So here we have, I

just wanted to add the state to the topology

that we're defining. And we're just adding,
all I

did was replace the logger one, and instead
of

logging, I'm calling partition persist, passing
in the factory.

I'm saying, I'm expecting. Now I'll just build
the

state. But I'm expecting a tuple that has
the

field message, and I'm passing use the basic
updater.

And that's it for getting a very basic topology

running. So the next step is let's go back

to our initial, like, Twitter hash tag example.
This

is somewhat what, like, this might be what
it

might look like. Ah, many mights. This is
what

it could look like as a Storm topology. You'd

start with a Tweet spout, however you decide
to

get your Tweets in. Via Redis, via directly
Twitter

API, whatever. You pass it, you'd pass it
to

a transform, which the entire, which its only
goal

is to get the Tweet body, get out the

hash tags and output them as tuples. The next

step is going to be an aggregate, and what

that's going to do is it's going to get

all the, it's going to get all the hash

tags and track how many counts there are.
And

then it's gonna send that to the state, and

the state is going to do the moving average

calculation.

This is what it might look like. Again, pretty

basic. Extract hash tags inherits from base
function. We

define execute and what it's gonna do is get

the Tweet body out of the Tuple, extract hash

tags. I believe it's the same code. And loop

over it and just emit new tuples.

Next, the aggregator. So the first thing it
does

is just init, so, the. Oh, yeah. First the,

an aggregate function is basically just like
Ruby's in,

inject on enumerable. So you pass it an initial

state and then it's going to loop over whatever

it is and pass, pass you the. So the

aggregate is the iteration. It's going to
pass in

the state that it's building up to as well

as each Tuple, and again, an optional output
stream

that you could output Tuples in mid-iteration.

And the init method, you return the initial
state.

Finally there's going to be an extra complete
method,

which is oh, we are done aggregating, so let

us then finally output our aggregation as
tuples to

the stream. And, in this case, just going
to

loop over our, our summary, which is a hash

map with a hash tag and a count. I'm

gonna output tuples that contain the hash
map, the

hash tag and the count.

And hook it all up. Pretty similar. The main

things with aggregates, aggregates functions
or aggregate transforms, you

want to call partition aggregate. And I will
cover

that in a bit. And also we're going to,

at the same time, add our trending topic state.

So this one might be implemented, let's again,
we

inherit from, OK, at the very all I do

is trending topic state inherits from state.
Exactly the

same as previously.

I'm going to implement begin commit and commit,
for

now. And I'm gonna leave them empty and come

back to it. And the update method is going

to take the hash tag and the count. And

it's going to do the exact same work that

it did in the Sidekiq one, for now. So

it's going to find an existing hash tag record

by name and update the moving average and
save

it again. And the updater is going to take

in all the tuples and just pass it to

the states.

So you might be wondering, well, I thought
streams

were unbounded sequences of tuples. So if
that is

the case, when does one call the complete
method?

Infinite time from now? I don't know. But
no.

What, the way Storm executes its topology
is it

executes in batches. So the way it works is

gonna be the Storm coordinator, is going to,
here

we go. Storm coordinator is going to tell
your

spout, yo spout. I'm gonna just read what
it

says. Yo spout, start batch id 123.

And the spout is going to then be like,

OK. Cool. We're starting batch 123. Let's
get some

data for it. I fetched two hundred tuples
from

whatever source. Like, I, I popped two hundred
messages

off of Redis. Seems good. And this is gonna

then go through everything we just saw, send
the

messages down. It's gonna go to the hash tag

function, which will then go to the aggregate
function,

which then go to states. The the states gonna

be, persist the result and send it back to

Storm.

So since an I can tell, we can tell

Storm, we completed batch 123. So what's,
so what

it means is, after, whenever there's a, a
part

of the API that appears to be time-based,
like

begin commit, commit, and in our case, the
complete

or the aggregate function, that's gonna be
called after

every single batch. So what, when we say aggregate

all, like, aggregate all the counts of the
hash

tags, we're really just doing it for a batch.

Oh yeah. And that it's going to start the

next batch. This just says start batch 124.

So is it web scale yet? Well, this is

obviously always the answer you have to ask
after

every single thing. Every single hour of work.

Well, this is what I've got so far. This

is what I told you was happening. We've got

a Tweet spout. It's sending data down a stream,

a single stream, into, to the extract hash
tags.

So we have one stream and, well, can a

single stream handle fifty thousand Tweets
per second? Well,

let's break it down for a bit.

This is the most basic topology. This is what

I said happened. This is what conceptually
it happens.

But, the way it executes is, a stream actually

has many partitions. So the number of partitions
is

configurable. But every single partition can,
is completely independent.

So it can run on completely different nodes
in

your Storm cluster. So if you have eight different

partitions, you're going, you have the opportunity
of read

both from the spout, read from eight different
servers

sending it, like, transforming and persisting
on eight different

servers. You could have up to, you can have

like a thousand different partitions if that's
how far

you need to go.

But the main point is that you can split

up the work load even though you have one

conceptual stream, you can split it up across
many

partitions and thus servers and threads. So
let's kind

of try to work through, again, the hash tag

thing through how it might work with partitions.
We

had the spout. That says RailsConf. And it
comes

out of the spout in one partition, and because

we haven't said anything yet, it's just gonna
say,

oh, we're just gonna send it to the aggregate

transform in the same partition. And it's
going to

go to the state as well on the same

partition.

Here's another bottom left, like, on the left
side,

another RailsConf tag. It's going to, it's
in a

completely different partition. So possibly
a different server. It's

going to go the aggregate in a completely
different

server, and ideally what we'd want is for,
at

the very end, we can do the moving average

calculation on the same server. So ideally
it would

end up on the same state partition.

But if we have another, like this is just

hash tag sleep, something I would like to
do,

it would stay on its own partition and it

would not, like, it doesn't need to be on

the, run on the same server that processes
the

RailsConf hash tag.

So in, but the one thing's like that, one

move from the aggregate middle partition up
to a

combining into one end point state partition,
we haven't

defined yet.

The way you do that is, at any point,

you can be, oh, at this point, I want

to partition my tuples by hash tag. And when

we say that, what we're saying is we're specifying

a field on the tuple, and we're saying, hey

storm, we would like, once we cross this point,

we would like any tuple that has the same

hash tag value to end up in the same

partition. So this, at this point, like, up
to

now, storm didn't, could actually run the
entire topology

and same partition on one server.

There was no need to, like, it might, I

mean you could specify, make it do that, but

there is actually no need to process, to have

a Tweet hit the network, basically. To go
to

a different node. The moment you add a partition,

like statement, you're saying, OK, yes, at
this point,

we're going to have to make sure that all

hash tags on the same value end up on

the same server. So it's going to, at this

point, try and, like, make sure to do that,

and thus may hit the network for it.

So how many partitions should we run? Well,
this

is very, very use-case specific. You could
run eight.

You could run five-hundred twelve. You, it,
it actually

depends on your use case. But in order to

answer this better, we need to talk a bit

about how Storm schedules partitions on the
cluster. So

first of all, there are many servers. And
you

have a cluster, you have many servers distributed,
yay.

So let's say, I don't know, three servers,
and

each server could have many threads per server,
because

it's very, very concurrent.

So you might have, I don't know, thirty, forty,

you, configurable amount of servers, threads
per the server.

And there can be many partitions per thread,
as

well. So every, a thread in Storm is considered

an executer. And an executer can have an arbitrary

number of partitions. So when you start up
your

cluster, like let's say you've got, you want
to

boot up a Storm cluster that is three nodes,

you have nine partitions, three threads. So
it works

out pretty nice. This is how it might, like,

start initially.

So you boot up and you have, like, anyway.

So it's going to assign three partitions per
thread,

per server. So that, the nice thing of dealing

with partitions, dealing with partitions is
actually a, something

that basically everything that ends up being
pretty distributed

does. Like, Reaq, Cassandra, Kafka, all these
things go

via partitions, and, because it's very nice.
But little

tangent.

So, what happens is, when you lose a server,

oh. We don't have these executers anymore.
We have

all these partitions, they need to be handled
still.

Storm is going to redistribute them across
existing servers.

So it's gonna be like, OK, all these, all

these partitions which represent work, work
load, are gonna

be moved to the other servers.

And, inversely, that's a little, on the top
is

a little server box, and I'm saying, ah, we're

adding a server to our cluster. And then we're

gonna rebalance it, and Storm is going to
take

the existing partitions and redistribute,
redistribute it across the

available servers.

Probably more evenly than my little graph
says, talks

ab- shows. OK.

It's
time for some real talk. Failure.

So, the question is not will there be failure.

The question is, how do we handle it? Cause,

if you're building a distributed system, at
some point

there is going to be fail, there is going

to be something that fails. And you can either

close your eyes and pretend it doesn't happen
or

you can think through the problems, like how
will

we recover? Is the system going to be in

an inconsistent state afterwards? Is, are
we gonna lose

availability during this time?

Handling failure properly is probably the
hardest part of

building any distributed system. But it's
not like you

have to use really fancy technologies to build
a

distributed system. Like if you build a plain-old
Rails

app to build a distributed system, you have
a,

the browser talks to the Rails app. And all

these problems that exist, maybe we don't
think about

it as much because we're OK with higher error

rates and inconsistencies. But, let's think,
like, even a

simple case like a signup form, like the user

fills up the signup form and hits submit and

it errors out, what happened, right? Did the,
did

it fail reaching our server? Did it reach
our

server and we started processing it and, but
we

had an exception? Did we, like, actually create
the

account in the database and the next time
the

user is going to refresh the page it'll add,

like, I'd like, I'd like to really signup.
I'm

gonna try to signup again.

And it's gonna be, oh no, you can't signup.

Your email is taken. Well, OK. How do we

handle all this? And there are ways. I'm just

kind of bringing up this simple case, which
I'm

sure everybody here has properly protected
their Rails app

against, because nobody in this room has ever
had

this bug happen.

But, I bring it up because this is a

relatively simple case, and it really just
gets more

complicated from here. So, for example, in
Storm, the

whole, like, it's fine. The whole point is
that,

you may have one input tuple. Like, one message

might be popped up from the queue, and as

you process it, you're going to end up generating

more tuples. Like, for example, we are going
to,

from one Tweet, we are going to extract many

hash tag messages. And if you, as you build

more and more complex topologies, this is
just going

to like, this can, like, fan out.

So what happens if, like, this one dies? So

everything else, like, succeeded, but that
one, and all

of its children, didn't. How do we, then,
what

do we do? Do we try to retry just

that one tuple? If we do that, we then

have to, well, we have to make, track, that

is that specific one that failed, that's gonna
involve

a lot of bookkeeping. The, and what happens
if

tuples have multiple parents? And one parent
ended up

succeeding but not the other?

And that's possible, with Storm. You can have
tuples

that have multiple parents because you can
have joins.

But how does one process it? Do you retry

the entire batch? It's hard.

I will get to how you handle it. But,

and the details of how Storm does it internally

are really interesting, and I would recommend
you go

and read the Wiki page on it, but I

don't have time to talk about it. So, going

back to the original Sidekiq thing, what happens
if

we're, like, halfway through the iteration
and we've persisted

some of them and, we, it just explodes halfway

through? Well, do we retry the message? Do
we,

I don't know. What do? What now?

Generally, we have to think about, well, when
we're

using a system like Sidekiq or Storm, what
are

the message processing guarantees? Because
it's going to change

how you work with it. Generally, the two,
the

two main ones are, are messages processed
at least

once, or are messages processed at most once?
And

you can only get one, not the other. Like

you can, I mean, yes, you can get exactly

one, see what I mean. If anyone says, ah,

we guarantee that a message is going to be

processed exactly once, they're lying.

It's impossible.

So to que, what we can try to do

is, well, Storm is at least once. But what

we can try to do is handle the case,

is like, can we handle, if we get messages

that show up multiple times, can we handle
that?

Can we maybe not redo the processing? Can
we,

like, can we try to ensure that we end

up with what, as close to as possible, as

close as possible to processing exactly once?

And the way Storm tries to approach this is

really pretty nice. Like I said, we have batches.

Everything's processed in batches. And what
Storm will guarantee

is that a batch, like the next batch will

not complete, like, will not go, reach completion
until

the previous one is fully committed. And it
will

also give monotonically increasing batch ids.

So it will guarantee that the next batch id

will have a bigger id than the previous one.

So we can combine this, and I will just

Tweet my slides and it will be easier to

read, but the main this is, I'm now introducing,

in the beginning, commit and running out of
time

so I'm gonna try to get through this.

But the main thing, if we, in the beginning

commit, we get the transaction id, what we
can

do is store it off, and what, and load

up the state and we will talk about that

later. But the important thing I want to reach

is, talk about it is here, in the update

function, we don't actually write to the database
anymore.

We just store that in memory. And the main

line is return if hash tag last transaction
id

equals transaction id. What that means is
we are

able to then see, OK, since we know the

guarantees, if for some reason a message failed
and

we reprocess the batch, we, and we know, we

can know that this hash tag actually passed
a

certain point, like passed that point, so
we don't

need to recompute the moving average on it.

And get as close as possible to exactly once.

So then, we store all this memory, and once

the commit happens, which happens at the end
of

the batch, we write all of this, we do

basically, we do all the work that hits the

outside world, which is right to the database.
We

do all that there. And if we reach completion,

the batch will complete successfully, and
we know, OK,

all the messages have been successfully processed.
But if

it fails during the persist, half of the them

will have the new transaction id and half
of

them will have the old transaction id, but
when

the batch gets replayed and we reload that,
we

will see, ah, this message already finished
this batch.

So to roll through this real quick, I know

it's, I believe I'm getting really close or
past

my time. But, so, yeah, let's get data. What

happens? Like, I fetch two hundred tuples.
Boom. We

go through. Aggregate. And it explodes.

Eventually, the Storm coordinator's gonna
be like, up above,

it says, guys, I did not get a completion

message. All right, fine. Spout, let, we're
going to

restart batch 123, because I did not get completion.

So the Tweet spout's gonna be like, OK, I

got this. I'm gonna re-emit exactly the same
two

hundred tuples in the same order that I did

before.

So what that means is if we successfully extract,

like, if we successfully implemented our transforms
to be

purely functional, as in completely deterministic
based off the

input, as we go through, we're gonna get exactly

the same outputs. Such that, once we get to

the state, I'd, in theory everything's going
to be

exactly the same, such that when we actually
run

the persist, we know that if something we
successfully

persisted, and we get the same transaction
id, we're

gonna be like, well, it's going to be exactly

the same output.

All right. So the main catch is, of course,

transforms must be purely functional, cause
otherwise this won't

work. That means not ever using time dot now

within your transforms. I actually changed
it on the

last slide, but I had to, like, zoom through

it. The main thing is, again, one, if you

have need to get the current time for a

batch, one way to do it is to have

a spout that only emits the current time.
And

every, if the batch gets re-emitted, it just
knows,

it saves off somewhere. For this batch id,
I

outputted this time.

And the next thing is that, it's going to

require the spouts to re-emit exactly identical
batches. Which,

unfortunately, I think most spouts do not.
I mean,

most, like Redis, does not support. So in
order

to get this level of guarantees, the only
queue

is Kafka. And I was, would like to talk

more about Kafka and why it's so awesome,
but

we can talk about that after. In general,
even

if you're not using Storm, Kafka is a really

amazing tool that I highly recommend.

So, TL;DR. Storm is a really powerful platform
for

writing workers. It's really great for stateful
jobs, and

by stateful jobs, I mean jobs that depend
on

the result of previous ones. It's good for
complex

data processing flows. And that's it. I think.
I

don't. My time wasn't up here, so I'm guessing

this is about right.

Am I really early or late? OK. Cool.

I didn't have a clock on either so I'm

like, guessing. It didn't start, I don't know.
All

right. Well, I'm done. Thanks. All right.