►
Description
Processing Billions of Events in Real-Time with Heron - Karthik Ramasamy, Twitter
Twitter generates tens of billions of events per hour when users interact with it. Analyzing these events to surface relevant content and to derive insights in real-time is a challenge. To address this, we developed and open sourced Heron, a new real time distributed streaming engine. In this presentation, we first describe the design goals of Heron and show how the Heron architecture achieves task isolation and resource reservation to ease debugging, troubleshooting, and seamless use of shared cluster infrastructure with other critical Twitter services. We subsequently explore how a topology self adjusts using back pressure so that the pace of the topology goes as its slowest component. Finally, we outline how Heron implements at-most-once and at-least-once semantics and we describe a few operational stories based on running Heron in production.
About Karthik Ramasamy
Karthik Ramasamy is the engineering manager and technical lead for real-time analytics at Twitter. He has two decades of experience working in parallel databases, big data infrastructure, and networking. He cofounded Locomatix, a company that specializes in real-time streaming processing on Hadoop and Cassandra using SQL, that was acquired by Twitter. Before Locomatix, he had a brief stint with Greenplum, where he worked on parallel query scheduling. Greenplum was eventually acquired by EMC for more than $300M. Prior to Greenplum, Karthik was at Juniper Networks, where he designed and delivered platforms, protocols, databases, and high availability solutions for network routers that are widely deployed on the internet. Before joining Juniper, at the University of Wisconsin he worked extensively in parallel database systems, query processing, scale out technologies, storage engines, and online analytical systems. Several of these research projects were later spun off as a company acquired by Teradata. Karthik is the author of several publications, patents, and Network Routing: Algorithms, Protocols and Architectures. He has a Ph.D. in computer science from the University of Wisconsin, Madison with a focus on databases.
Join us for KubeCon + CloudNativeCon in Barcelona May 20 - 23, Shanghai June 24 - 26, and San Diego November 18 - 21! Learn more at https://kubecon.io. The conference features presentations from developers and end users of Kubernetes, Prometheus, Envoy and all of the other CNCF-hosted projects.
A
Okay,
thanks
for
coming
so
the
title
of
my
talk
is
be
processing,
billions
of
events
in
real
time
using
the
newly
released
open
source
software
called
a
Twitter
henan
twitter
has
parent,
has
been
in
production
with
him
to
death
for
more
than
two
one
half
years,
and
we
just
got
it
open
source
like
a
few
months
ago
and
already
we're
hitting
around
probably
thousand
pull
request
so
far,
I
think
we
should
be
crossing
at
any
time
next
week.
I
guess
so
anyway.
A
Detection
signed
a
few
more
things
like
that,
and
so
I
also
teach
at
Berkeley
some
of
the
undergrads
es
courses,
in
addition
to
bring
some
little
bit
of
research
but
other
than
that
like
so
then,
before
coming
to
tutor,
I
was
running
my
own
company
that
got
scooped
like
the
scooped
up
by
Twitter
and
we
have
been
I've
been
doing
real
time
for
probably
around
eight
eight
years
on
now.
Okay,
so
yeah
with
that
introduction.
A
How
do
you
go
to
the
top
of
the
queue
always
then
I
will
follow
it
by
conclusion,
so
to
give
a
heron
road
view,
how
many
of
you
have
known
storm,
oh
nice?
So
so,
essentially,
we
can
think
about
her
own
as
a
next
hydration
of
storm
accept
the
fact
that
architecture
Lee
it
has
been
changed
drastically
underneath
and
but
the
terminology
and
the
API
we
have
kept
the
same,
but
the
flexibility
is
that
we
can
add
a
new
API
is
on
top
very
quickly.
A
If
we
need
to,
but
the
architecture
is
sound
enough,
that
the
code
does
not
have
to
change,
but
anyway,
I
will
come
to
that
later.
So
to
give
us
terminology
about
the
storm,
/
heron,
so
your
topology
is
it's
a
real
time
job
and
it's
nothing,
but
that
dag
and
there
are
two
type
of
vertices
in
the
dag
and
there
are
as
one
daya
one
type
of
what
is
called
the
spout.
A
So
the
other
one
is
called
bowls,
and
these
vertices
represent
some
kind
of
computations
on
the
edges
within
the
diag
represents
the
streams
of
data
that
is
flowing
between
these
vertices
and
the
spout
essentially
is
a
type
of
vertices
which
kind
of
taps
into
the
data
source
and
injects
the
data
into
the
topology
situationally.
The
real
time
job
and
the
bolt
essentially
are
processing
elements
in
the
sense
like
it
takes
incoming
topple
transforms
into
whatever
way
it
needs
to
be
done
and
emits
an
outgoing
trouble,
and
some
examples
include
filtering
aggregation.
A
So
with
this
terminology
like,
let
us
look
how
a
dag
looks
like.
So
this
is
a
typical
dag,
so
you
have
the
couple
of
spouts,
which
in
turn
feeling
into
a
next
stage
bowls
which
are
like
a
three
of
them,
both
one
through
both
three
and
once
they're
done
with
the
part
of
the
Cross
thing.
Then
it
goes
to
the
next
stage,
bowls,
which
is
the
bold
for,
and
bold
five
under
the
end
stage.
Bowls
also
capable
of
writing
the
results
into
another
system.
A
So
that's
a
so
one
idea
was
that
we
could
have
introduced
the
notion
of
a
sinks
like
instead
of
a
bowl,
and
we
could
have
spout
bolt
and
a
sink
but
say
we
wanted
to
kind
of
reduce
the
amount
of
new
term
knowledge
that
we
use.
So
instead
we
just
decide
to
keep
the
bold
term,
but
the
bolt
can
do
some
kind
of
arbitrary
processing
so
which
means
they
can
overlap
processing
as
well
as
storage
the
results
into
third
party
system
simultaneously.
If
that
is,
would
like
to
do
so.
A
To
give
a
concrete
example,
let
us
take
a
word
count,
topology,
and
in
this
case
what
you
are
trying
to
do
is
you're,
going
to
count
the
number
of
distinct
words
on
a
live
stream
of
real-time
tweets
that
are
coming
at
you.
So,
in
order
to
do
that,
you
need
to
have
some
kind
of
a
tweet
spout
that
in
turn
taps
into
the
to
terrify
holes
where
you
can
use
the
Twitter
API
to
grab
those
tweets,
and
once
that
is
the
tweets
have
been
tapped.
So
the
tweet
has
to
be
broken
up
into
words
somewhere.
A
So
that's
why
you
introduce
this
parts
to
eat,
bold,
where
the
tweet
coming
out
of
the
spout
is
given
to
the
first
tweet
bowl,
which
in
turn
breaks
up
into
words.
Then
you,
after
the
words,
are
broken.
The
words
have
to
be
given
to.
There
was
some
kind
of
a
bolt
called
vodka
bowl
which
actually
constitutes
a
logical
client,
and
so
this
is
essentially
a
dag.
A
In
the
database
parlance
I
mean
how
many
of
you
taken
the
intro
databases
or
grad
level
database
courses,
there's
a
notion
of
a
logical
plan
in
the
physical
plant
before
the
plan
gets
executed
in
the
physical
system.
So
we
essentially
claim
this
as
a
more
of
a
logical
plan.
But
thing
is
like
because
of
the
sheer
size
of
data
that
you
might
be
getting.
You
may
not
be
able
to
handle
that
in
a
single
process.
A
Let
us
say,
like
the
tweet
spout
and
a
part
street
board
and
a
word
count
bold
or
a
single
process
of
why
each
then
you
may
not
be
able
to
handle
the
brunt
of
data
in
a
single
process.
That's
one
possibility
another
possibility.
Even
if
you
have
multiple
process
within
a
single
machine,
they
may
not
be
able
to
handle
all
the
data
that
you
get
as
well,
so,
which
means
your
physical
execution
has
to
be
very
different
than
what
you
look
at
in
the
logical
plan.
A
So
so
in
reality,
like
the
physical
plant,
looks
like
this,
where
you
have
multiple
number
of
tweets,
pout
task
and
multiple
number
of
parts
to
eat
both
task
and
the
multiple
number
of
word
count
both
task,
and
this
is
what
we
call
is
the
parallelism.
So
you
can
give
any
number
of
parallelism
for
each
one
of
those
stages,
so
you
can
say
that
I
want,
like
15
since
of
spout
to
eat,
spout
and
I,
want
hundred
instances
of
power.
A
Sweet,
bold
and
200
instance
of
word
count
both
depending
upon
how
much
CPU
how
much
memory
they
need
and
all
the
various
things
right.
So
this
is
Ashley
worth
of
what
you
call
the
physical
plan
when
it
gets
realized
in
the
execution
stage.
Now
this
individual
tasks
are
packed
in
some
base
under
distributed
into
a
big
cluster
so
that
they
can
discover
each
other
and
start
exchanging
data
from
not
only
from
a
within
a
single
machine,
but
across
machines
as
well.
So
I
will
go
into
those
things
a
little
bit
detail.
A
So
now
one
of
the
challenges
here
is
like
given
apart
apart,
Street
bolts,
image,
said
a
data
item
or
a
tuple
which
of
the
word
count
bolt.
It
should
go
to
because
you
have
multiple
tasks
right,
so
that
is
a.
How
do
we
do
it
so
like
there
is
something
called
notion
of
body
callers
stream
groupings,
and
the
first
is
called
the
shuffle
grouping
where
it
can
pick
randomly
one
of
those
bowls
downstream
bowls
and
send
it
to
it.
A
Then
there
is
something
called
feels
grouping
where
you
take
the
trouble
and
examine
some
of
the
attributes
of
the
topple
and
based
on
the
contents
of
those
attributes.
Either
you
do
some
kind
of
a
hash
function
or
whatever
it
might
be,
and
accordingly
direct
the
couple
to
that
appropriate
downstream
bolt.
Then
the
third
one
is
something
called
all
grouping
where
you
can
replicate
to
everybody
and
similarly,
like
you,
have
something
called
global
grouping
where
you
have
multiple
tasks:
feeding
off
to
one
single
guys.
A
This
is
used
for
debugging
and
printing
and
all
the
various
stuff,
but
ninety-nine
percent
of
the
time
you
have
a
combination
of
shuffle
grouping
and
fields.
Grouping
is
what
we
end
up
using
so
so
now.
Going
back
to
the
word
count
apology:
you
can
say
that
by
the
way
like
a
tweet
spout,
if
it
has
to
send
the
data
to
bar
sweet
Bowl,
shuffle
grouping
will
suffice,
because
anybody
can
parse
that
break
that
tweet
into
words,
but
on
the
other
hand,
when
you
go
from
afar
street
ball
to
a
word
count
board.
A
A
A
You
cannot
go
beyond
that
right,
so
it
couldn't
enforce
that,
and
because
of
that,
like
you
used
to
Kapali
multiple
topologies
that
were
running
and
sharing
some
missions,
they
used
to
trample
all
over
the
place,
and
that
leads
to
a
lot
of
challenging
issues
that
used
to
happen
because,
like
doing
especially
the
events
time
I
mean
tutor
is
known
for
real
time.
So
during
events
time
like
Oscar
and
Super
Bowl.
So
those
events
there
are
traffic
spikes
up
quite
a
bit,
and
because
of
that,
like
everybody,
starts
occupying
more
cpus.
A
A
It's
very
hard
to
debug
what
heck
is
going
on,
so
that
is-
and
second
aspect
is
like
you-
cannot
even
do
a
heap
dump
or
some
kind
of
what
do
you
call
the
attaching
your
kid
kind
of
tool
so
that
we
can
debug
order
performance
profile
that
spout
or
the
bolt
or
whatever,
so
that
you
can
identify
your
perform
bottleneck?
So
then
the
third
one
is
the
use
of
manageability.
So
we
wanted
to
make
sure
that
the
instead
of
running
storm
on
a
separate
cluster
on
its
own.
A
We
wanted
to
go
to
multi-ton
cluster
because,
as
a
small
team,
we
have
to
support
the
entire
Twitter's
needs
right.
That
means
like
around
30
or
40
teams
who
are
using
the
software
or
the
infra.
So
what
we
have
to
do
is
to
make
sure
that
we
don't
have
to
maintain
a
separate
cluster
and
oh
the
overhead.
That
goes
along
with
it,
in
fact
like
when
we
were
maintaining
the
same
cluster.
A
The
it
took
around
the
two
to
three
weeks
to
take
a
topology
that
was
in
development,
going
into
production
because
of
the
fact
that
we
are
to
go
through
some
kind
of
a
cap
planning
to
find
out
how
much
those
missions
are
required
for
the
topology,
and
we
ought
to
kind
of
send
it
to
the
cap,
planning
or
the
hardware
department
and
which,
in
turn
released
the
hardware.
Then
we
or
do
then
somebody
has
to
manually
add
those
missions
into
the
cluster
or
data
center.
A
Then
it
has
to
be
then
some
kind
of
entry
has
to
be
put
in
so
that
when
the
topology
kind
of
a
describes
order
tells
ok.
This
is
the
topology.
Ok
you're,
entitled
to
only
do
this
number
of
machines
right
so
because
of
that,
like
it
took
like
three
weeks
to
two
weeks
in
order
to
get
a
topology
to
into
production,
which
is
unacceptable
so
that
so
so
what
we
did
is
like
how
to
make
it
easier
and
run
it
in
a
multi-time
cluster.
A
There
are
a
couple
of
frameworks
that
are,
and
on
top
of
it,
which
in
turn
generates
those
topologies
automatically
so
the
whole
machinery
we
can
lift
it
by
changing
the
API,
so
the
is
given
is
you
have
a
AP
a
compatible?
The
second
aspect
is
like
a
task.
Isolation.
As
I
said,
we
want
isolation
at
all
levels.
One
is
at
the
individual
task
level.
A
Remember
the
past
week
bolt
and
the
word
count
ask
each
one
should
be
having
isolation
of
its
own
and
similarly
like
when
you
pack
a
bunch
of
these
processes
into
on
container
kind
of
stuff.
Then
we
need
isolation
of
the
container
level.
Then
the
individual
topology
should
have
isolation
so
that
they
don't
share
any
kind
of
state
with
the
they're
more
so
that
it
is
even
killing
off
a
topology
does
not
affect
anybody
else.
A
So
then,
finally,
like
we
decided
like
we
used
mainstream
languages
like
C++,
Java
and
Python,
compared
to
some
of
the
functional
languages
that
storm
was
employing
in
well,
they
wrote
the
initial
version,
which
is
something
called
closure.
I,
don't
know
that
you
have
we
heard
about
the
language
closure,
so
it
was
written
enclosure
and
the
closure
and
interaction
with
JVM
leads
to
lot
of
j
GC
shoes.
A
And
then
we
decide
what
k
we
under
the
tools
for
closure
was
not
that
great
so,
which
means
as
a
developer
productivity
is
not
that
fast
enough
to
deliver
any
new
stuff
and
even
fixing
bugs
and
all
so.
Instead,
we'll
stick
with
mainstream
languages
so
that
we
can
get
enough
programmers
so
so
with
that
are
all
those
decent
ations.
The
heron
architecture
looks
very
simple:
I
mean
the
one
of
the
systems
principle
is
keep
it
simple.
So
that's
what
we
did
like.
A
So
we
got
away
from
the
fact
that
a
we
don't
want
anymore
schedulers,
because
already
the
lot
of
communities
among
run
in
schedulers,
that
is
working,
Miso's,
yawn
and
Cuban
attics.
So
there
is
enough
momentum
behind
the
scheduler
coming.
It
is
so
it
reinventing
the
wheel
for
doing
another
scheduler,
just
not
in
our
doesn't
make
any
sense.
So
instead
we
piggyback
on
existing
scheduler
that
could
you
be
scheduled,
it
could
be
anything
and
as
long
as
some
that
are,
some
mapping
is
needed
to
map
hair
and
into
that
scheduler.
A
Then
we
are
okay,
so
which
means
when
you
submit
a
topology,
it
looks
like
at
another
job
to
the
scheduler
and
the
scheduler
in
turn
schedules
the
various
topology
based
on
the
resource
requirements
of
that
apology.
So
that's
what
it
is
so
now,
let's
drill
down
detail
into
more
how
they
knew
your
topology
looks
like
so
we
so
right
from
the
start.
We
exploited
the
container,
it's
quite
about
the
to
the
container
level.
We
isolated
several
containers
and
the
at
the
high
level.
A
There
are
two
different
kind
of
containers:
one
is
the
master
container
and
another
one
is
called
data
containers,
so
the
ones
that
you
see
in
the
bottom
or
data
containers
and
the
one
at
the
top
where
the
topology
master
is
running,
is
called
the
master
container.
So
right
from
the
start,
we
took
advantage
of
the
container
architecture.
A
So
when
a
topology
gets
launched
in
the
first
place,
the
master
container
is
the
first
container
that
gets
started
and
the
master
container
contains
something
called
the
topology
master,
which
is
the
first
process
that
comes
up
and
it
looks
at
the
dag
and
how
much
it
sources
it
needs,
and
everything
and
also
runs
something
called
packing
algorithm
so,
which
means
like
hey
I,
need
total
of
ten
containers.
How
do
I
pack,
these
parts
treat
both
tasks
and
the
word
count.
A
Bowl
task
and
the
tweet
spout
pass
into
one
into
multiple
containers,
and
so
that
everybody
knows
where
that
container
they
are
running
on
so
so
that
is
called
the
packing
algorithm.
So
once
the
packing
is
done,
then
the
topology
master
in
turn
will
request
the
scheduler
hey
by
the
way
I
needed
10
containers,
and
each
of
the
container
should
be
of
four
cpus
and
probably
like,
say,
a
couple
of
gig
of
memory.
A
It
advertises
itself
saying
that,
hey
by
the
way,
if
anybody
up
my
topology
wants
to
reach
out
to
me,
this
is
the
host
and
port
that
I
am
listening
on.
So
that
location
is
a
returning.
The
zookeeper
cluster,
which
acts
as
a
synchronization
mechanism.
Then,
once
everybody,
the
data
containers
and
the
corresponding
stream
manager
process
comes
up,
the
stream
manager
goes
and
looks
in
the
zookeeper
cluster,
hey
by
the
way
various
met,
apology,
master.
A
So
the
moment
it
discovers
where
my
topology
master
is
it
contacts
the
topology
master,
hey
by
the
way,
I
am
container
X
and
I'm
ready,
so
like
a
kind
of
a
hollow
message
and
registers
itself,
and
so
the
topology
master
waits
until
every
container
that
has
been
spawned
has
been
reported
in
and
once
everybody
reports
in
then
it
constitutes
something
called
a
physical
plant.
I
mean
the
logical
plan
is
the
diag
which
is
already
available
on
the
zookeeper.
A
So,
which
means
some
parts
of
the
depaul
you'll
be
running
on
a
data
container
one
and
some
parts
will
be
running
on
data
container,
two
and
so
on,
and
because
of
that,
like
the
data
has
to
be
moved
around
and
in
order
to
move
around,
the
stream
managers
have
to
discover
each
other.
You
can
think
about.
Team
manager
is
more
of
a
sophisticated
data,
router
info
on
some
form
where
it
has
to
route
the
data
to
appropriate
containers
so
once
the
steam,
and
it
has
to
discover
each
other.
A
So
that's
why
it
forms
the
top
olly
murs
reforms
a
physical
plant
and
the
moment
the
physical
plant
is
formed,
and
it's
towards
the
zookeeper
claw
that
physical
plant
into
the
zookeeper
cluster
as
well,
and
then
it
kind
of
broadcast
that
physical
plant
to
all
the
data
containers
are
the
stream
managers.
So
once
everybody
gets
third
physical
plan,
they
know
the
roles
in
the
sense
like
remember.
The
the
stream
manager
will
connect
to
each
other
right
and
also
the
instances
so
which
is
officially
the
very.
A
So
we
made
it
each
one,
the
task
as
an
individual
process
on
each
one
so
that
you
can
connect
to
it
and
query
its
memory
or
you
can
keep
dump
it
or
you
can
even
do
a
performance
profiling
of
the
task
without
affecting
anybody
else,
and
also
you
can
do
fine,
grained
resource
allocation
as
well,
because
when
you
have
a
one
single
process,
how
much
resource
you
want
to
allocate.
It
is
unknown
to
you
unless
you
know
what
are
the
threads
and
other
things
are
mapping
which
is
controlled
by
packing
algorithms.
A
In
a
here
you
can
tell
like
it.
This
process
requires
the
smudge
and,
depending
upon
how
the
packing
is
goes
up,
you
do
light
up
to
the
overall
container
resource,
limited
resource
that
you
need
for
the
container,
and
so
so
so
now,
once
the
physical
plan
is
given
to
all
the
stream
managers
and
they
know
how
to
connect
to
each
other,
so
they
will
form
a.
What
do
you
call
a
fully
connected
graph
so
that
they
can
start
exchanging
data?
So
so
that's
how
it
starts
executing
now.
A
Let
us
look
at
the
physic
failure
scenarios,
because
it's
all
running
in
a
distributed
system
process
failure,
machine
failures
are
pretty
common.
So
now,
let's
take
the
example
of
the
master
container
failure,
so
when
the
master
container
fails,
there's
nothing
surf,
acting
because
master
container
is
more
on
the
control
path,
not
in
the
data
path.
So
when
the
master
container
dies,
so
the
data
will
continue.
A
The
container
failure
I
mean
another
option,
is
you
could
run
multiple
must
master
containers
with
the
kind
of
a
master-slave
relationship
which
can
be
elected
by
a
zookeeper.
Also,
what
thing
is
in
practice
that
we
found
out
as
long
as
we
can
quickly?
Scheduler
another
container,
you're
fine
I
mean
in
practice.
We
have
not
seen
any
moisture
content
of
failure
affecting
and
cost
you
a
lot
of
incidents.
So
so
we
kind
of
stuck
with
it.
So
now
let
us
look
at
the
failure
of
a
data
container.
A
So
whenever
data
container
fails
to
paly,
Marshall
will
come
to
know
because
there's
a
keepalive
messages
that
goes
from
the
stream
manager
to
the
topology
master.
So
the
moment
that
the
connection
goes
down
or
the
keeper
lives
do
not
come
any
more
then
tow,
poly
master
knows
that
the
container
is
dead.
So
then
another
container
gets
spawned
off
by
the
scheduler
or
based
on
the
request,
form
topology
master.
So
then
that
container
will
come
up
and
the
first
order
of
business
is
rediscover
where
my
topology
master
is
and
send
my
host
and
port.
A
Then,
once
the
poly
must
forget
said
it
constitutes
a
new
physical
plant,
because
one
guy
has
changed
because
everybody
has
to
connect
to
this
guy,
then
it
saves
it
into
the
again.
The
zookeeper
was
a
physical
plant
state.
Then
all
the
got
all
the
guys
who
fail
the
connection
to
that
the
world
guy
who
died.
We
connect
to
the
new
guy.
A
Again,
data
will
start
exchange
and
during
the
time
there
is
some
amount
of
buffering
going
on,
because
if
the
container
is
average-sized
because
in
a
24
core
mission,
the
average
size
is
somewhere
around
4
260.
So
if
you
do
ask
after
mission
order
three
quarter
or
a
full
machine,
it's
hard
for
the
scheduler
to
get
that
mission
quickly,
because
the
time
to
allocate
becomes
longer
the
more
the
resources
that
he
asked
for
an
upper
container
basis.
A
If
it's
an
average
sized
container,
we
won't
even
see
a
glitch
in
the
failure
in
the
recomp
and
the
coming
of
back
of
the
container
and
all
the
physical
plan
and
all
the
reasoning
it's
pretty
quick.
We
probably
will
clocked
it
with
a
less
than
20
seconds
or
so
so
any
questions
so
far
on
this
topology
architecture,
yes
mm-hmm,
so
it
is.
The
stream
manager
has
a
direct
connection
to
the
poly
master
using
a
TCP
connection,
and
that
connection
is
a
control,
pod
connection
and
then
periodically.
A
It
sends
a
hello
messages,
that's
it
that
is
hardly
like,
because
it
doesn't
every
few
seconds.
So
that
is
a
configurable
element.
You
can
control
it.
How
much
you
want
to
because
of
you
awfully!
Obviously,
if
the
more
the
interval
that
you
configure
the
more
that
failure
deduction
will
go
so
it's
a
thing,
I
trade
off
the
do
to
play
with,
but
that
traffic
is
very
small.
So,
yes,.
A
Service
discovery,
so
the
zookeeper
is
a
tionally
uses
a
service
discovery,
for
example,
for
everybody
I.
All
the
containers
want
to
discover
the
topology
master,
where
it
is
running
the
topology
masters
advertise
to
itself
on
the
zookeeper
and
the
zookeeper
us
a
tree
like
hierarchy
right
and
when
one
of
the
hierarchy
it
will
have
a
topology
name.
Since
everybody
knows
what
apology
name
that
they
are
running
on
you
keen
on
that
part,
then
you
know
whether
to
poly
master
is
yes,
yes,
mm-hmm.
A
A
A
Actually,
then,
I
cannot
give
out
the
actual
numbers,
but
yeah,
like
some
of
the
jobs,
varies
from
one
stage
to
two
stage:
10
stages,
the
one
thing
that
we
did
is
after
moving
from
storm
to
hair
and
one
of
the
biggest
and
things
that
we
got
is
we
got
a
3x
reduction,
in
course,
and
memory.
So
we
made
the
system
so
efficient
in
terms
of
moving
data
and
all
the
various
talk,
so
we
saw
a
huge
reduction.
In
course,
I
mean
we
were
running
around
around
3500
nodes
or
something
around
4000
nodes
and
storm.
A
We
brought
it
down
to
less
than
thousand
five
nodes
like
so
that
way
like
it's
a
huge
savings
from
capex,
as
well
as
the
operational
costs
of
it.
So
so
that
was
one
of
the
highlights
of
this
thing,
and
so
some
of
the
use
cases
involves
broadly
like
a
real
time.
Etl
I
mean
for
extract,
transform
and
load
so
where
the
data
is
transformed
and
loaded
into
some
other
destination
systems.
A
The
real
time
bi
I
mean
VI,
is
recently
like
where
you
have
one
of
the
application
is
like
when
people
are
interacting
with
Twitter
from
a
website
or
even
mobile
phones
or
different
countries,
so
you
can
continuously
view
them
at
a
second
granularity
where
the
traffic
is
spiking.
We
said
coming
from
UK
or
it's
coming
from
Belarus
and
all
the
various
things
and
also
broken
down
by
what
operating
system
people
use
an
iphone
or
vs
android
based
on
region.
A
So
you
can
see
the
entire
breakdown
of
these
various
dimensions
we
used
as
a
twitter
command
center,
where
you
keep
wanting
monitoring
all
these
crafts.
So
that's
pretty
cool
and
we
also
do
a
lot
of
spam
detection
product
abuse,
all
the
various
things
also
on
top
on
top
of
her
and
where
they
build
some
kind
of
a
machine
learning
model
and
all
the
various
things
and
the
acid
data
coming
in.
They
used
to
check
against
that
model.
Whether
there
is
any
violation
or
not
so
that
is
all
runs
on
top
of
her
and
the
trend.
A
Computations,
the
trending
hashtag
trends
and
all
the
various
trends
that
suckers
they're
all
competed
on
top
of
her
own
as
well,
and
in
addition
to
that,
some
of
the
ML
model
buildings
and
model
enhancements
and
model.
All
the
things
also
learn
on
top
of
heaven
and,
as
I
said,
media
like
videos
and
pictures
if
they
want
to
classify
based
on
certain
things,
they
also
pick
up
those
media
and
extract
the
features
and
classify
those
features
based
on
whatever
the
algorithms
that
you
want
to
do.
A
And
third,
one
is
a
it
also
being
heavily
used
for
ops,
where
the
data
from
the
missions
and
the
bigger
clusters
are
continuously
being
collected,
and
we
have
proactively
look
for
if
there
is
an
emission
failure
is
out
the
impending
machine
failures,
or
so.
These
are
the
broad
some
in
categories
of
uses,
so,
like
I,
wanted
to
highlight
a
couple
of
things
that
we
did
the
hair
and
reasonably
right.
A
So,
as
you
all
know,
the
big
data
system
is
content
really
evolving
in
the
sense
like
a
tomorrow,
the
new
scheduler
comes
up
and
then
the
comment
that
community
goes
away.
Then
another
scheduler
comes
up
similarly
like
today
how
to
boost
in
fashion
and
tomorrow.
Something
else
will
be
in
fashion,
because
how
do
please
being
rewritten
by
Facebook
in
C++,
because
they
want
a
high
performance
or
whatever
right,
if
they're
open
sources
that
becomes
stranded.
So
it's
in
the
constantly
changing
environment
of
several
components.
A
A
Okay,
so,
like
we
I
bet
that
the
original
paper
and
con
mock
I
mean,
if
you
remember
this
from
it,
used
to
be
from
CMU
from
early
90s,
so
I
said
like
they
took
the
monolithic
kernel
operating
system
and
stripped
it
down
and
say
that
okay,
this
is
the
basic
thing
that
we
need.
The
rest
of
them
can
be
run
as
a
process
like
even
the
pages,
and
the
virtual
memory
was
running
as
pages
as
long
as
the
colonel
has
the
basic
IPC
and
the
scheduling
and
other
functionality.
So
I
said
like.
A
Why
can't
we
do
the
same
thing
here?
So
if
you
look
at
the
streaming
systems,
they
are
nothing
but
data.
Movers
I
mean
they
move
data
quickly
as
fast
as
can
and
a
scheduler
computation,
and
this
computation
is
moved
to
the
next
stages,
so
which
means
the
basic
in
inter
our
infra
IPC
is
a
constant
is
needed,
and
also
like
what
regular
scheduler
that
you
are
drawn
on,
because
schedule
is
also
changing
continuously
how's,
the
resource
requested
and
the
scheduling
and
monitoring
those
containers
and
all
the
various
things.
A
So
the
abstract
allowed
to
do
what
we
call
as
a
service
provider
interval
like
a
regular
API.
We
invented
something
called
spi
with
the
service
provider
interface
and
similarly,
the
state
manager,
which
is
the
state
synchronization
point
similar
to
zookeeper.
That
also
be,
is
pluggable,
because
today
you
might
be
using
zookeeper
tomorrow,
you've
somebody
will
write
based
on
something
draft.
Then
they
might
have
a
different
consensus
algorithm.
So
we
want
her
to
abstract
that
or
not
that
also
out
so
so
we
have
a
state
manager
spi
under
scheduler
spi
under
the
basic
thing.
A
So
now
all
of
them
became
a
process
on
top
of
it,
so
which
means
I
well
established
protocol
using
TCP
and
some
kind
of
product
of
based
protocols.
Everybody
is
interacting
in
a
different
way:
I
mean
through
thy
pcs,
so
if
the
body
must
interacts
with
team
manager,
there's
a
well
established
protocol
as
long
as
you
confirm
with
this
protocol.
You're
fine.
A
So
the
solo
also
allows
a
lot
of
flexibility
where
the
polyesters
are
able
to
map
quickly
to
me
so
slasher
order,
which
is
our
production
environment,
and
we
are
able
to
quickly
max
what
was
able
to
move
that
to
run
on
yon
very
quickly
within
just
two
or
three
weeks
and
then
HPC
community,
which
is
on
a
high
performance
community.
They
use
a
scheduler
called
the
slurm
scheduler.
A
They
were
able
to
move
in
a
couple
of
weeks
as
well
and
then,
after
that,
the
some
people
asked
for
me
so
/
marathon,
okay,
that's
also
is
fine,
so
so
like
the
scheduler,
writing
became
so
easy
so
and
it
kind
of
proved
itself
based
on
various
scheduler
mappings
and
so
then
also
like
going
to
a
more
process
like
architecture.
So,
for
example,
the
instance
that
I
mentioned
where
your
tasks
are
actually
running.
So
what
happens?
A
If
we
want
to
do
tomorrow
right
the
your
topologies
in
Python
right
so
then
I
can
write
a
Python
instant
that
goes
along
with
the
API.
That
instance,
will
none
the
native
Python
code
in
a
Python
vm.
So
we
can
write
the
instance
in
Python
without
affecting
anybody
else.
That's
what
we
did
for
byte
an
API
and
some
guys
at
Adobe
wanted
to
write
a
C++.
Topology
I
said:
go
for
it
as
long
as
he
conformed
to
this
protocol
of
instant.
A
In
addition
to
that,
they
also
wanted
our
non
GC
environment,
where
it's
running
predictively
right
so
because
of
those
two
main
reason
they
want
to
write
it
in
C++,
but
so
the
architecture
allows
it
and
tomorrow,
if
I
wanted
to
change
the
stream
manager
to
work
on
top
of
InfiniBand
or
even
working
on
a
shared
memory,
kind
of
architecture
yeah,
you
can
just
spin
up,
write
a
difference,
team
manager
and
register.
It
then
automatically
will
spin
it
up.
A
So
that
way,
this
gave
a
very
flexible
architecture
and
the
environment,
a
kind
of
surrounding
it
is
continuously
changing.
So
so
we
published,
we
are
finished
a
paper
on
this
and
submitted
to
ICD
international
conference
on
data
engineering.
It
should
be
available
anytime,
like
probably
early
in
the
first
quarter
of
so
anyway,
so
the
plug
and
play
component
as
environment
changes
code
does
not
change
and
so
multi-language
instances.
A
As
I
mentioned,
you
can
write
Python
in
C++
and
all
the
various
stuff,
and
you
can
change
the
stream
manager
depending
upon
the
transport
or
even
crossing
semantics,
at
least
once
at
most
one's
exactly
ones
all
kind
of
semantics.
And
finally,
this
one
of
the
another
important
thing
is:
it
buys
us
a
lot
of
velocity
like
the
so
the
guy
who's
developing
stream
manager
word
says
the
guy
who's
developing
instant.
They
can
run
off
independently
and
go
out
the
faster
velocity
and
they
contested
in
their
own
environment
and
everything
before
they
decide.
A
Okay,
this
is
ready
to
go
out,
so
there's
no
highly
intertwined
this
thing,
so
that
helps
a
lot.
Even
if
you
break
one
thing
doesn't
mean
that
the
other
thing
is
broken
right,
so
so
I
mean
like
so
it
actually
runs
on
their
our
development
environment.
We
have
this
notion
of
a
local
scheduler
which
we
wrote
so
that
it
can
run
on
a
single
laptop,
and
so
we
have
the
notion
of
an
uploader
where
the
jars
and
other
job
execution
code
has
to
be
uploaded
right.
A
So
that
also
uses
a
local
file
system
and
that's
our
primary
development
environment
and
the
miso
/
otherwise,
testing
also
we
use
zookeeper
and
we
tested
with
the
HDFS
uploaders
very
upload,
the
job
test
sheer
face.
Then
we
have
this.
The
production
environment
is
purely
in
Aurora.
The
zookeeper
is
the
state
manager
and
the
packet
uploader,
which
allows
you
to
do
versioning
management.
Where
you
hear
a
job,
they
are
uploading
several
times.
It
keeps
a
very
a
version
of
it
so
that,
if
something
happens
in
the
new
deploy,
you
can
roll
back.
A
So
those
kind
of
things
you
can
do
so
now
I'm
going
to
go
into
what
we
call
like.
So,
even
though
it
is
returning
this
micro
kernel,
like
a
kind
of
architecture,
did
we
sacrifice
performance
or
what
so?
That
is
why
I
wanted
to
bring
this
up
like.
Even
though
we
went
to
a
more
of
a
process
model
as
compared
to
a
thread
model,
did
we
lose
performance
because
of
that
or
we
are
able
to
gain?
So
this
is
what
it
gives
them
more
insight
about
what
actually
the
Costas.
A
Then
that's
what
emitted
out
of
the
even
spouts,
the
spout
is
doing
all
this
logic
of
absorbing
later
filtering,
as
well
as
expanding
the
topple
and
sending
it
out,
and
the
aggregate
both
takes
the
incoming
tuples
and
it
aggregates
based
on
every
second,
then
outputs
into
red
is
and
that
our
rate
is
around
25
to
40
2
million
topple
per
minute,
and
so
this
is
the
overall.
The
resource
consumed
was
like
a
radius.
A
We
dedicated
a
machine
for
reading,
which
is
24
course,
and
but
the
actual
quotes
that
were
used
was
around
two
to
four
and,
of
course
it
admission
content
of
48
gb
now
with
Haren,
which
is
running
on
the
multi-tenant
cluster,
the
we
requested
course
of
120
codes.
That
is
approximately
around
five
machines,
and
the
actual
you
say
that
we
found
was
around
30
to
50
cores
and
the
memory
requested
was
200,
but
actually
we
use
just
only
180
gb.
A
Now
when
you
break
down
where
the
heck,
the
whole
usage
of
the
course
is
the
course
are
going.
So
we
found
out
like
this
was
kind
of
an
intuitive
the
spout
instances
where
the
guys
were
taking
a
lot
of
time
like
around
eighty
four
percent
of
the
whole
CPUs
out
of
the
30
to
50
core
usage.
It's
eighty-four
percent
went
into
spouts
and
the
bolts
just
took
around
nine
percent,
and
the
Heron
is
overhead.
A
A
I
mean
it's
deeply
nested
from
around
six
to
seven
levels
of
nesting
and,
of
course,
we
need
to
have
or
some
kind
of
a
way
for
some
kind
of
data
evolution
or
a
schema,
the
evolution
so
that
we
don't
break
down
all
the
application
that
were
returned
with
the
older
schema
right.
So
we
that
is
a
necessary
thing,
but
so
the
beers
kind
of
appalled
to
see
that
the
deceleration
cost
is
the
major
component
of
it.
The
next
aspect
is
like
getting
there
they're
out
of
Kafka.
A
So
the
next
problem
that
I
wanted
to
do
is
like
how?
What
do
you
do
especially
remember,
you're,
running
the
whole
dag
in
a
distributed
fashion?
Right?
What
happens
if
one
of
the
guys
going
slow
I
mean?
So
this
is
what
typically,
we
call
the
stragglers.
So
the
stragglers
are
a
norm
in
the
distributed
system.
We
got
to
deal
with
it
because
they
can
account
for
different
reasons,
and
the
first
reason
is
a
bad
host.
I
mean
in
a
hundred
thousand
cluster
like
some
bad
apples
will
be
always
coming
up
and
going
so.
A
But
the
thing
is,
since
the
streaming
is
something
that
you
cannot
stop.
It
keeps
on
going
continuously
because
remember
to
the
topology,
once
you
launch
it
there's
no
way
to
stop
it
unless
you
kill
it.
So
it's
so.
It's
not
like
any
interactive
query
system,
very
type,
a
query
and
the
query
finishes
and
the
results
come
through,
whereas
the
here
results
are
continuously
being
generated
as
long
as
the
data
is,
they
have
to
be
generated,
but
it's
the
ball.
You
will
still
run
if
there
is
no
data
as
well.
A
So
the
first
problem
is
a
bad
host.
So
if
you're
unfortunate
that
you
got
scheduled
in
a
bad
host,
what
do
you
do
then?
The
second
one
is
executions.
Q
I
mean
twitter
is
known
for.
Ask
you
because
of
the
fact
that
the
asymmetric
nature
of
the
user
and
the
followers
right.
So
what
do
you
do
with
the
executions
q?
Then?
If
the
topology
is
not
adequately
provision,
then
again,
stragglers
could
come
in
because
you're
not
able
to
process
then
coming
data
rate
right.
A
So
so,
let's
look
at
what
other
approaches
that
we
do
in
order
to
handle
strugglers.
So
one
is
by
the
way
the
guy
is
not
receiving
as
fast
as
I
am
sending
okay,
so
I'll
just
drop
it,
so
that
is
rationally,
send
it
to
a
straggler
drop
data.
So
so
then
another
hand.
The
second
approach
is
hey:
I
will
slow
down
to
the
pace
of
what
the
standards
are
absorbing
the
data
from
me,
so
everybody
slows
down.
A
So
that
is
the
second
approach,
and
this
is
very
typical
of
a
tcp/ip
where
you
have
senders
and
receivers
and
when
Sanders
starts
with
very
fast,
then
if
the
moment,
the
rate
at
which
the
receivers
are
sending
a
problem
and
then
the
readies
a
window
size
like
it,
tcp,
slow
start
and
all
the
various
things
right
and
they
said
alone
on
a
speed
and
the
third
one
is
detective
ease,
ask
stragglers
and
reschedule
them.
So
that's
the
third
approach
so
like
what
are
the
pros
and
cons
of
that
so
drop
data
strategy.
A
One
of
the
biggest
issues
unpredictable.
You
have
no
idea
where
the
data
is
being
dropped,
especially
in
a
complex
drag
of
like
running
on
500
600
containers
right.
So
you
don't
have
enough.
It
affects
accuracy
because,
since
you
don't
know
where
is
being
dropped,
hey
by
the
way,
I'm
supposed
to
a
get
a
count
of
like
a
couple
of
million
but
suddenly
I'm
getting
something
like
a
500,000
which
is
not
correct,
but
why?
It
is
not
correct.
A
So,
which
means
we
don't
have
visibility
into
what
is
causing
the
issue
right,
because
a
lot
of
things
could
be
happening.
So,
on
the
other
hand,
the
slowdown,
Center
strategy,
it's
kind
of
provides
predictability
because,
as
everybody
slows
down
for
the
slowest
guy,
the
whole
thing
just
put
gets
pushed
into
the
source.
That
source
is
start
lagging
where
the
starting
point
of
the
pointer
at
which
you
write
the
data
versus
the
pointer
that
you're
reading
the
data
it
starts
diverging
because
of
the
fact
that
they
are
not
absorbing
the
data
as
fast
as
I.
A
When
a
touchdown
happens
boom
like
a
little
like
34
exit,
let's
go
in
spite
then
again
will
come
down
like
its
last
for
a
few
seconds.
That's
it
then
again,
like
some
bad
call
of
vampire
everybody
boom
will
go
up
again,
come
down
so
like
I,
since
Lee,
like
the
spikes,
are
a
kind
of
norm
within
the
to
test
data
sets
that
you
get,
and
so
this
one
handle
spikes
also
nicely.
A
So,
let's
add
lemon,
so
we
event
went
with
the
slowdown
center
strategy
and
that
we
started
implementing
that
and
so
the
way
implement
it
is
like.
Let
me
give
an
example
an
illustrated
by
an
example.
So
let
us
take
this
linear
topology,
where
yo
a
single
spout
and
that
in
turn
goes
to
a
bolt
b2,
which
in
turns
feeds
to
a
bolt
b3,
and
finally,
it
goes
to
bowl
p,
for
which
does
some
additional
processing
and
storing
the
results.
A
So
so
this
is
the
how
the
assuming
three
instance
of
each
one
of
them
each
one
of
the
vertices
in
the
node.
Let
us
say
it
spread
around
four
containers.
As
you
know,
like
all
the
stream
managers,
are
there
fully
connected
graph
right
under
exchanging
data?
Now,
let
us
say:
take
the
example
of
b2
going
slow,
because
v2
is
not
absorbing
data
from
the
other
guys
and
I
mean
from
s
ones
and
other
things
as
fast
as
it
should.
So.
What
happens?
Is
the
stream
manager
on
that
container
will
realize?
A
Because
that
is
a
guy
who's,
pushing
the
data
to
the
guy
be
too
it
will
realize
a
bother
with
this.
Guy
is
not
going
as
fast
as
it
because
my
buffers
are
overflowing
because
team
manager
is
the
one
which
manages
the
buffer,
and
so
then
it
sends
a
message
hey
by
the
way,
some
guy.
In
my
manager,
in
in
my
container,
is
experiencing
back
pressure,
so
I'm
going
to
initiate
a
back
pressure
protocol
saying
that
hey
by
the
way
everybody
initiate
whatever
you
need
to
do
for
back
pressure,
please
go
ahead
and
do
it.
A
So
once
the
buffers
in
the
stream
energy
drain
of
below
a
certain
threshold,
then
we
again
sender
what
he
called
really
back
pressure.
Then
the
spouts
are
open
and
keep
going
so
so
that
is
a
very
simple
mechanism
and
which
all
which
allows
you
to
go
in
to
initiate
and
relieve
back
pressure
quickly
and
also
there
is
a
cushion
in
the
sense
like
you
have
these
buffers
in
the
stream
manager,
and
you
can
pick
up
the
threshold
on
the
Lolich.
A
What
you
call
a
low-water
mark
and
high
water
mark
and
accordingly
said
those
values
and
those
values
in
turn
will
govern
when
the
initiate
back
pressure
occurs
and
the
lead
back
pressure
occurs.
So
if
there
is
a
huge
gap
between
the
two
the
initiate
and
the
leave
will
settle
down
those
smooth
curve
rather
than
having
something
like
like
initiate
and
relieve,
initiate
and
the
live
like
a
flip-flop
kind
of
thing
right.
So
so,
yes,
a
deep
in
the
sense
like
how
big,
so
they
typically
sounded.
Meg.
A
Yes,
I,
don't
Meg,
but
that's
again,
configurable!
So,
depending
upon
your
application
requirements,
everything
you
can,
you
can
configure
on
a
per
topology
level
also,
so
so
so
that
seems
to
work
well
in
practice,
as
you
can,
I
mean
if
we
look
at
the
lot,
so
logical
topology.
This
is
like
15
containers.
A
It
knows
that
that
container
is
a
cause
of
the
root
cause
of
the
bad
pressure,
but
sometimes
the
back
pressure
will
most
of
the
time
the
back
pressure
will
recover
itself
automatically
and
there
are
scenarios
in
which
it
doesn't
recover
at
all
because
of
hi
GC.
Something
is
happening
so
then
manual
intervention
will
be
required.
So
we
are
working
on
some
kind
of
anomaly:
detection,
algorithms,
where
so
automatically
we
look
at
all
these
various
pout
and
other
instances
and
the
moment
something
is
behaving
anomalous
and
we
look
at
the
correlations
and
all
the
various
things.
A
Then,
if
there
is
high
degree
of
confidence
that
that
container
is
experiencing
slow
down,
then
automatically
you
should
on
the
container
more
to
the
next
host
or
whatever
it
is
so
like
the
back
pressure
in
practice
most
scenarios
it
requires,
as
I've
mentioned,
without
any
manual
intervention,
which
means
no
pager
incidents.
So
the
sustained
back
pressure
occurs
in
some
places,
irrecoverable,
GC
cycles
or
a
bad
or
faulty
hose.
A
Then,
finally,
like
sometimes
use
a
preferred
dropping
of
data,
the
care
orderly,
the
latest
data,
so
we
also
run
what
we
call
for
detecting
and
bad
and
faulty
hose.
We
also
run
a
separate
heron
topology,
which
takes
the
data
and
continuously
does
some
kind
of
simple
machine
learning
to
figure
out
what
is
going
on
and
now
the
we
are
not
automatically
taking
those
hosts
out
of
a
scheduler.
Instead,
it
goes
through
a
manual
process
where
it's
given
to
the
appropriate
administrator.
They
even
toned
go
manually,
remove
it,
but
we
will
make
it
more
automatic.
A
Also
so
finally,
like
this
is
a
load
shedding
so
sampling
based
approaches,
but
doesn't
work
very
well
because
it's
very
hard
to
reason
about
sampling
across
uniform,
distributed
spots,
where
everybody's
reading
in
a
different
rates
and
everything
right
and
we
cannot
up
sample
the
results
and
the
opp
sampling
results,
might
not
work
for
something
like
a
median
mode
and
all
the
staff
too.
So
we
simply
went
with
the
drop
based
approaches,
simply
drop
the
older
data.
A
The
spout
lacks
a
threshold
and
takes
a
lag
threshold
and
the
Lark
adjustments,
and
this
seems
to
work
in
practice,
because
here
people
in
this
case
is
people
don't
care
about
accuracy.
Instead,
they
want
to
be
on
top
of
the
real
time
so
because
they
might
be
trending
computing,
some
trends
or
whatever
it
is.
So,
even
if,
for
example,
in
the
cases
where
we
have
seen
some
rock
outages
right
so
which
mean
the
data
does
not
come
for
four
hours.
A
So
when
the
data
comes
after
four
hours,
then
all
the
topologies
who
are
want
to
process
the
data
they
will
process,
of
course,
but
they
are
out.
Apologies
where
hey
I,
don't
care
about
what
has
happened
in
the
last
for
us
I
want
to
know
what
is
right
now
so
move
my
pointers
directly.
Do
then
we
have
the
latest
data
is
coming
from,
so
that
works.
A
So
if
you
want
like
there,
we
have
these
three
papers
returned
already
and
it's
all
available
on
the
web,
and
if
you
wanted
to
learn
about
it,
feel
free
to
learn
about
it.
And
if
you
have
any
questions,
please
post
them
in
herrin
users.
Mailing
list,
so
we'll
be
happy
to
answer.
So
if
an
earnest
open-sourced,
if
you
guys
want
to
use
it,
go
for
it,
one
thing
that
we
Canada
is
reliability,
which
is
because
we
are,
after
hair
into
production
the
next
reduction
number
of
incidents.
A
So
the
new
active
work
that
we
are
just
wrapping
up
is
what
you
call
a
scaling
automatically
the
topology
can
scale
and
we
are
pretty
much
done
at
various
and
testing,
and
initially
the
rollout
will
be
on
manual
scaling
where
you
do
a
hair
and
update
command,
then
we'll
automatically
expand
the
job
from
like
a
10
containers,
200
containers
without
pulling
down
the
existing
job,
and
that
will
become
more
automatic
in
q1.
So,
yes,
that's
all
I
had
any
questions.