►
Description
Join us for Kubernetes Forums Seoul, Sydney, Bengaluru and Delhi - learn more at kubecon.io
Don't miss KubeCon + CloudNativeCon 2020 events in Amsterdam March 30 - April 2, Shanghai July 28-30 and Boston November 17-20! Learn more at kubecon.io. The conference features presentations from developers and end users of Kubernetes, Prometheus, Envoy, and all of the other CNCF-hosted projects
Scaling and Securing Spark on Kubernetes at Bloomberg - Ilan Filonenko, Bloomberg
In the management of its Data Science Platform, Bloomberg has always focused on providing tenants with secure, reliable, and scalable solutions for their machine learning workflows and ETL pipelines. In adapting Kubernetes to support a diverse set of machine learning workloads, we decided to also support Apache Spark with Native Kubernetes integration. In this talk we'll discuss how we designed: a scalable and resilient External Shuffle Service for Dynamic Resource Allocation, a pluggable interface for secure worker creation, and a token renewal service that handles privacy and security across Spark jobs. These topics will address multi-tenancy, data security and privacy, and elastic resource scalability in the context of running Spark natively on Kubernetes, with an emphasis on disaggregated compute.
https://sched.co/MPal
A
Just
in
case,
you
guys
were
wondering
these
slides
are
posted,
so
you
don't
take
photos
of
the
slides
just
future
reference.
It's
easier
that
way.
Okay,
so
I'll
get
started.
Thank
you
all
for
coming:
I'm,
Milan,
Fallon,
anko
and
I'm,
a
software
engineer
at
Bloomberg
working
on
the
data
science
infrastructure
team
and
today
we'll
be
talking
about
scaling,
securing
spark
on
kubernetes
at
Bloomberg.
A
So
machine
learning
workloads
come
in
many
shapes
and
sizes,
but
the
challenges
that
ml
teams
face
are
quite
common,
even
in
established
enterprises.
Some
of
these
concerns
include
data,
permissioning
scheduling
and
Rito's
monitoring
for
tasks
that
include
feature
generation,
model,
training
and
model
development
in
an
effort
to
solve
some
of
these
challenges,
specifically
regarding
ml
questions
and
problems,
Bloomberg
has
developed
a
versatile
data
science
platform
that
provides
sophisticated
compute
environment
for
engineers
to
manage
their
ETL
pipelines
and
they're
modeled
with
model
development
life
cycles.
A
One
integral
component
of
most
ETL
pipelines
is
Apache
spark,
so
just
as
a
show
of
reference,
how
many
people
know
of
Apache
spark
in
this
room?
Okay,
awesome,
that's
really
great
and
how
many
of
you
actually
use
it
within
your
kubernetes
environments?
Okay,
even
better.
Now
we're
gonna
discuss
that
a
little
bit
so
just
for
those
who
didn't
raise
their
hand,
spark
functions
as
a
robust
framework
that
uses
a
general
execution
engine
to
provide
distributed.
Data
processing
and
one
of
Sparx
powerful
applications
in
fee
is
feature
transformation
and
generation.
A
A
The
spark
context
then
can
connect
to
several
types
of
cluster
managers,
which
includes
standalone,
meso,
CR
or
core
kubernetes,
and
these
cluster
managers,
in
essence,
just
allocate
resources
across
applications
now
upon
receiving
Pro
provision.
Executors
Sparkman
sends
your
application
code
to
the
executors
and
then,
after
which
the
spark
context
would
then
send
tasks
to
to
the
executors
to
run
that
the
executor
will
then
use
a
variety
of
caching
strategies
for
various
SPARC
files
that
are
traditionally
stored
on
the
local
file
system.
A
So
that
would
be,
for
example,
that's
displayed
over
here,
and
persistence
is
also
traditionally
done
to
local
HDFS
data
nodes
for
stuff,
like
hive
tables,
we
just
orderly
here.
The
same
time,
executors
might
need
to
communicate
with
each
other
in
order
to
read
from
their
other
local
caches.
Now,
storage
relating
to
the
data
that
you
wish
to
process
is
also
traditionally
co-located
for
a
data
locality,
speed.
Ups.
As
such,
you
can
see
how
the
storage
and
compute
are
on
the
same
node,
thereby
co-located.
A
However,
this
design
is
constraining
when
we
try
to
extend
kubernetes
as
a
cluster
manager
for
SPARC,
which
we
will
elaborate
on
later,
but
also
it
isn't
consistent
with
the
with
the
industry's
trend
towards
promoting
disaggregated
compute.
But
what
is
this
aggregated,
compute
or
Disick?
So
this
AGG
is
the
cluster
separation
of
storage
and
compute
and
as
Bloomberg's
technical
problems
grow
in
scope.
So
does
our
data
and
unsurprisingly,
compute
nodes
that
aren't
designed
for
large
disk
space
can't
handle
certain
workloads
of
SPARC
is
doing
its
file?
A
Caching
by
writing
too
slow
to
its
local
file
system
and,
unsurprisingly,
compute
nodes.
Aren't
designed
in
this
way
will
now
be
started,
taking
advantage
of
the
hardware
that
you
can
now
specialize.
So,
for
example,
your
compute
nodes
will
now
be
RAM
heavy,
while
your
storage
nodes
can
be
disk.
Adding
this
aggregated
compute
will
also
allow
your
cluster
management
and
heart
and
hardware
upgrade
cycles
to
be
done
separately,
especially
within
Blumberg,
where
we
have
specialized
teams
that
manage
storage.
This
is
particularly
appealing.
A
So
Bloomberg
has
incorporated
this
dis,
Agra
proach
in
the
design
of
its
data
science
platform,
and
what
you
can
immediately
notice
is
the
separation
between
storage
and
compute.
Our
users
store
their
data
in
s3,
HDFS
or
some
other
data
stores
of
their
choosing
and
one
goal
in
building
this
platform
is
to
make
it
easy
to
access
data
and
various
data
sources.
So,
on
the
bottom
right
we
have
our
compute
cluster,
which
is
made
up
of
many
CPU
and
GPU
nodes
which
all
which
has
all
bits,
resource,
scheduling
and
management
done
via
kubernetes.
A
A
Now,
let's
dig,
let's
dig
into
the
Eternals
of
the
job
the
platform
job,
so
the
life
cycle
looks
something
like
this
was
traditional
for,
like
an
operator
controller
architecture,
so
user
submits
an
API
resource
to
the
kubernetes
api
server
in
a
declarative
format
like
Y
Amal,
then
there's
a
platform
controller
which
watches
for
changes
to
a
resource
of
interest.
It's
worth
to
note
that
this
controller
runs
in
a
privileged
namespace.
A
The
controller
then
acts
like
a
state
machine
by
reading
the
updates
that
happen
in
an
appropriate
way
by
looking
at
the
current
state
compared
to
the
current
to
the
desired
state
of
the
world.
Now,
because
our
controller
runs
in
a
privileged
namespace,
it
can
now
submit
relevant
changes
and
requests
back
to
the
API
server.
So
this
could
come
in
the
form
of
a
pod
resource.
A
For
example,
the
API
server
will
then
schedule
the
pods
appropriately
and,
as
such,
the
SPARC
job
can
now
be
thought
also
as
a
platform
job,
thereby
the
controller's
responsible
for
requesting
resources
needed
to
spin
up
a
spark
class
spark
cluster.
Now
you
guys
probably
seen
this
diagram
a
lot
of
times,
but
let's
talk
a
little
bit
about
how
this
kind
of
fits
into
the
kubernetes
world.
A
So
the
in
the
initial
days
that
of
the
data
science
platform,
we
decided
to
launch
spark
via
stand-alone,
which
means
that
the
driver
communicates
with
a
master
pod
as
opposed
to
the
kubernetes
api
server
directly
as
such,
by
using
an
in
in
container,
the
driver
would
wait
until
the
master
and
worker
nodes
are
provisioned
and
reachable
before
launch
in
a
spark
application
defined
in
the
job
spec.
It
is
worth
to
note
that
our
storage
nodes
are
actually
separated
which
contain
most
the
user
application
data
from
the
from
a
computer.
A
However,
the
cache
layer,
as
you
can
see
here,
is
still
sitting
on
our
compute
nodes.
Now
some
drawbacks
become
very
apparent,
the
first
of
which
is
that
you're
unable
to
elastically
scale
within
a
spark
application
as
there's
no
direct
communication
with
the
API
server.
The
second
was
that
the
master,
node
and
worker
knows
needed
to
be
set
up
before
launching
the
spark
application.
A
So
if
you
guys
use
a
lot
of
Jupiter,
this
is
particularly
problematic
when
trying
to
create
a
spark
context
within
Jupiter
as
a
as
well
B
require
that
you
pre
launch
a
headless
spark
job
that
you
need
to
interact
with
now
this.
Traditionally,
we
are
scaling
up
our
user
account.
We
found
that
this
lets
a
lot
of
problems
with
the
resource
management,
as
sometimes
these
properties
clusters
weren't
properly
reap
reaps
upon
notebook
termination.
A
So
to
explain
that
better,
pretty
much
users
wouldn't
users
who
would
launch
a
spark
job
within
their
Jupiter
notebook
wouldn't,
would
need
to
be
responsible
to
terminate
that
same
spark
job
and,
as
you
can
imagine,
that's
not
always
the
case
so
to
solve
this
Bloomberg
actively
work
to
introduce
Cabrini's
native
integration
into
spark,
2
3,
+
2
4.
To
solve
these
very
issues.
Now
we
can
see
the
architecture
suddenly
change
by
having
direct
by
having
the
driver
directly
communicate
with
the
API
server.
A
So
as
such,
you
can
now
think
of
the
driver
as
a
controller
in
the
architecture
of
the
date
of
the
data
science
platform.
So
this
is
the
current
state
of
where
the
community
is
at
with
in
terms
of
spark
and
kubernetes.
However,
there
were
issues
when
we
started
integrating
this
with
in
our
production,
Bloomberg
environment.
A
Therefore,
to
have
a
driver
function
as
a
controller
is
not
possible
within
our
environment
because
it's
creating
pods.
Secondly,
the
vast
array
of
data
sources
that
Bloomberg
uses
are
all
secured
and
the
process
of
data
retrieval
within
our
compute
jobs
lack
to
manage
identity
service
that
integrates
with
Bloomberg
jobs,
especially
the
SPARC
and
lastly,
Sparx
caching
on
local
disk
makes
enable
a
fully
disaggregated
spark
environment
impossible.
Just
go
back
a
tiny,
but
this
cash
handling
here,
because
it's
being
wrote
to
the
local
file
system,
restricts
us
from
doing
a
fully
disaggregated
environment.
A
So,
let's
start
with
security,
as
we
said
earlier,
the
ability
for
pods
to
be
created
from
the
users
of
namespaces
forbidden,
but
even
if
it
were
allowed
via
some
form
of
restricting
restricted
web
hooks,
these
web
hooks
will
be
applied
to
all
pods,
including
system
pods,
which
could
bring
a
nose
will
slow
down.
Furthermore,
there's
no
ability
to
do
an
all
or
nothing
pod
creation,
we're
like
all
pods
required
for
a
spark
cluster
would
wait
to
be
registered
with
the
API
server
until
the
reaches
quota
is
available.
A
This
use
case
is
important
for
different
resource
scheduling
scenarios,
where
you
would
wish
to
tune
when
to
create
a
batch
of
executors
based
on
the
cluster
load.
So
this
customization
would
solve
some
complications
that
could
arise
with
regards
to
resource
deadlocks
between
competing
spark
jobs
on
a
single
workspace.
A
So
Bloomberg
developed
a
solution
by
extending
an
API
that
we
developed
that
hooks
into
the
kubernetes
cluster
scheduler
back
end
and
this
API.
This
API
is
default.
Implementation
extends
the
current
logic
which
we
just
launched
pods,
but
internally
we
extend
this
API
to
create
an
update,
a
custom
resource
that
we
call
the
exec
pod
scalar,
but
users
for
this
API.
A
If
you,
the
committee,
wants,
the
user
can
extend
any
form
of
custom
resource
that
they
choose,
so
so,
let's
jump
into
an
example,
spark
job
and
to
see
how
we
leverage
the
exec
pod
scalar
custom
resource
definition.
So
the
user
who
lives
in
the
users
namespace.
He
would
begin
by
submitting
a
spark
job
to
the
API
server,
using
some
declarative
format
like
Hamill.
A
For
example,
a
controller
living
in
a
privileged
namespace
would
watch
for
custom
resources
and
upon
seeing
a
request
for
a
spark
job
with
provision
and
Driver
pod,
with
all
the
appropriate
web
hook
and
modifications
and
upon
there
being
enough
resources
in
the
users
quota
for
their
driver
and
the
executors
within
the
controller
would
then
launch
a
driver
pod
now
by
measuring
the
quota
needed
for
the
driver
and
executors.
This
allows
for
us
to
avoid
hanging
drivers
who
are
deadlocked
from
launching
as
computers
when
competing
spark.
A
Jobs
are
all
running
in
the
same
instance,
which
is
pretty
important
because
there's
a
lot
of
situations
where
you
might
have
a
heavy
cluster
load
and
drivers
that
are
just
waiting.
So
you
now
have
the
overpopulation
of
too
many
drivers.
The
driver
pod
will
now
run
the
spark
application
and
upon
it,
requesting
executors
and
its
back-end.
It
would
create
an
exact
pod,
scalar
custom
resource,
which
the
controller
will
also
be
watching
for
upon
seeing
requests
for
an
exact
pod,
scalar
custom
resource.
The
controller
were
then
provision.
A
The
executors,
with
the
appropriate
web
hooks
and
modifications
upon
seen
again
the
appropriate
resource
quota
being
available
for
the
request
and
executors.
This
is
the
handle
of
race
condition
in
which
multiple
SPARC
jobs
could
have
been
started
after
the
period
in
which
the
driver
plus
executors
resource
count
was
available.
To
note
all
executors
are
created
at
once
upon
the
availability
of
resources,
as
opposed
to
having
only
a
portion
of
the
executors
created
and
subsequently
blocking
simultaneous
spark
jobs
who
might
be
in
the
same
deadlock
scenario.
A
So,
as
you
can
see
now,
the
driver
does
not
create
the
executors,
which
would
then
give
the
ability
for
users
to
cut
to
customize
their
own
pods,
but
instead
the
driver
will
setting
the
executors
pods
that
are
to
be
created
and
by
filtering
on
unique
spark
application.
Pods
the
driver
will
then
send
the
spark
application
information,
such
as
the
application,
specs
tasks
to
the
executors.
A
So
after
solving
one
of
the
challenges
with
compute
security,
another
one
else
presented
was
with
relation
to
storage
security.
Bloomberg
has
a
variety
of
data
sources
that
he
uses
for
its
model
development,
each
of
which
are
highly
secured
and
to
simplify
the
experience
of
using
moods
at
the
data
science
platform
when
interacting
with,
for
example,
kerberized,
HDFS
or
s3
the
data
science
platform
designed
to
manage
token
service
for
all
jobs.
A
The
token
service
is
serviceable
via
the
simple
inclusion
of
the
user
identity,
name
that
is
specify
on
your
job
specification
and
one
of
the
features
that
it
automatically
handles:
Hadoop
token
of
renewal
and
retrieval
for
HDFS
high
of
HBase
and
the
like.
Now
by
putting
the
token
service
behind
the
controller,
this
allows
for
reusability
and
standardization
across
a
variety
of
jobs,
including
spark
and
also
another
popular
job
that
uses
HDFS
data,
which
is
tensorflow.
A
So
let's
see
how
it
works.
So
the
controller
from
an
admin
namespace
communicates
out
to
a
variety
of
data
sources
and
management
tools,
which
in
sparks
case
includes
HDFS
and
s3.
The
user
begins
by
creating
an
identity,
custom
resource
and
a
controller,
will
then
read
the
custom
resource
and
have
the
users
privileged
credentials
stored
as
a
secret
for
jobs
in
the
users
namespace
to
when
a
job
is
looking
for
a
particular
identity.
The
controller
would
then
handle
bootstrapping
the
job
with
that
at
any
resource.
It
automatically
handles
updating,
expiring
tokens
when
dealing
with
long-running
jobs.
A
Now
these
long-running
jobs
are
quite
common
in
Bloomberg's
ecosystem,
as
I'm
sure
it
is
in
yours
when
we
deal
with
spark
ml
as
well
spark
streaming
jobs
that
could
run
for
weeks.
So
how
will
that
update?
Look
in
the
world
of
secure
HDFS?
Well,
the
controller
within
function
as
a
proxy
user
on
behalf
of
the
job
user
and
using
the
controller's
own
principle,
it
would
receive
an
HDFS
service
ticket
by
communicating
with
the
Kerberos
node
next
we'll
use
that
ticket
to
receive
a
delegation.
A
Token,
that's
bounded
to
the
job
users
principle,
to
ensure
that
the
job
user
only
is
actually
the
job
user
data,
and
we
also
provide
additional
security
if
I
put
in
a
token
and
the
secret
only
privy
to
that
user.
Therefore,
the
expire
token
would
be
updated
in
the
secret.
The
driver
and
executors
based
on
kubernetes
will
automatically
detect,
updates
to
the
secrets
and
have
their
tokens
renewed.
As
a
result,
however,
despite
detecting
updates
to
the
token
file
to
hook
into
spark,
we
added
an
additional
feature
to
spark
a
doob
utils.
A
It
allows
for
the
driver
upon
changes
to
the
token
secret,
to
send
token
updates
to
the
executors
to
then
kick
off
the
same
UGI
process
that
is
consistent
across
all
resource
managers.
So,
in
addition,
we
saw
that
in
other
jobs
like
tensorflow,
which
lactose
had
a
beauty
on
the
soju
butyl
UGI
feature,
we
needed
to
support
renewal
to
be
done
via
delegation
tokens,
and
so
you
propose
a
patch
to
Hadoop
core
to
support
this.
A
So
the
reason
for
this
is
the
following:
on
the
implicit,
the
implicit
assumption
in
the
old
mutation
was
that
yarn
functions
as
a
scheduling
agent,
with
access
to
appropriate
key
tabs
for
the
renewal
of
Kerberos
tickets
and
delegation
tokens.
However,
jobs
I
would
interact
with
kerberized
Hadoop
services
like
HDFS,
hiver,
HBase
and
use
an
external
schedule
like
communities.
Typically,
don't
have
access
to
key
tabs.
A
So
let's
go
a
little
to
the
spark
internals
to
allow
for
us
to
understand
where
they're,
strict,
where
the
restrictions
exist,
stopping
us
from
enabling
to
zag.
So
let's
define
an
example
user
story,
so
you
have
your
example:
spark
application.
So,
let's
you
have
a
large
data
set
a
few
petabytes
in
size,
for
example,
and
you
wish
to
run
a
couple
simple
transformations
on
it,
like
adding
+1
to
your
values
and
a
few
more
complex
operations
like
sort,
for
example,
before
collecting
the
results
into
the
driver.
A
So
spark
leverage
is
an
operator
graph
for
users
to
allow
for
them
to
apply
a
variety
of
transformations
that
won't
get
sent
to
the
dag
scheduler
until
an
action
like
collect,
for
example,
is
performed.
So
in
this
diagram
we
show
how
operations
are
broken
up
into
stages,
which
stage
can
contain
tasks,
abused
tasks
based
on
data
partitions
now
there.
So
there
could
be
many
tasks
Piatt
pipeline
together
in
a
single
stage
and
in
the
end
the
stages
are
passed
into
the
task.
Scheduler.
A
What
you
said
earlier
is
responsible
for
distributing
the
task
to
executors,
so
you
pay
attention
to
the
space
in
between
stage
and
and
n
plus
one.
You
can
see
an
alt
all
communication
that
is
called
a
shuffle
operation,
so
this
alt
all
communication
is
critical
for
operations
like
sorting,
which
is
done
by
your
example.
Spark
application,
above
as
I
said
earlier
in
the
user
story,
so
in
technical
terms,
shuffles,
when
you
move
data
from
one
stage
to
another,
while
partitioning
your
data,
so
the
reducers
receive
their
respective
partitions.
A
So
there
are
two
specific
important
things
to
note
about.
The
following
shuffle
example
illustrate
the
first
of
which
is
that
SPARC
implements
shuffle
using
files
on
disk.
Now
this
was
designed
with
a
strong
fault.
Tolerance
guarantees
as
well
as
a
simple
reasoning
that
not
all
shuffle
data
could
necessarily
fit
in
memory
which,
as
you
can
imagine
in
the
use
case,
I
just
gave
with
petabytes
in
size
just
cut,
that's
quite
common.
A
The
second
is
that
in
stage
n,
we
have
four
mappers
for
the
four
partitions
and
in
stage
n
plus
one
we
have
three
reducers
for
the
three
of
resulting
partitions,
so
this
serves
as
a
valuable
introduction
to
some
concept
called
dynamic
resource
allocation,
which
is,
when
you
have
a
spark
application,
miss
giving
back
resources
to
the
cluster.
If
they
no
longer
need
those
resources,
they
can
also
request
them
back
again
later
when
they
needed
based
on
their
demands.
As
you
can
see,
we
could
actually
give
back
a
work.
A
Resource
went
after
completing
stage
and,
as
we
moved
to
stage
n
+1,
this
is
valuable
as
executors
that
aren't
given
tasks
should
be
repurpose
for
other
spark
jobs.
So,
in
the
case
that
we
want
to
rescale
our
executor
talent,
we
need
to
ensure
that
these
shuffle
files
aren't
lost
when
executors
are
scaled
down,
and
if
these
shuffle
files
are
available
for
newly
created
executors.
A
Now,
what
kind
of
files
are
used
by
SPARC
to
enable
shuffle
the
first
kind
of
temporary
file
that
SPARC
uses
are
spill
files?
These
files
are
created
when
a
task
running
on
an
executor
spills
to
disk
after
being
unable
to
store
it
in
memory,
and,
as
you
can
imagine,
this
can
happen
quite
often
as
many
spill
files
can
be
created
on
a
given
stage.
The
next
kind
of
file
that
can
be
created
as
a
shuffle
file
which,
as
I
said
earlier,
are
created
whenever
you
want
to
do
a
shuffle
operation
between
stages.
A
This
file
is
created
by
the
mapper
via
shuffle,
right,
which
the
reducer
would
then
read
via
shuffle
fetch,
but
only
parts
of
partitions
and
the
case
that
spill
files
do
exist.
The
mapper
will
then
merge
all
the
Spill
files
that
happen
in
the
map
stage
and
put
that
merge
content
into
the
shuffle
file
for
later
stages.
To
then
have
so
in
the
current
implementation
of
shuffle.
Shuffle
files
involve
two
files
when
it
comes
to
shuffle
fetches
by
later
stages,
the
first
of
which
are
the
they
are.
A
The
data
files
which
are
written
by
the
mapper
and
the
second
is
an
index
file
which
are
purply
indexes
where,
in
the
data
file,
the
partition
begins
and
ends
so
for
reading
shuffle
bytes,
you
would
first
read
the
index
file,
determine
wearing
the
data
file.
You
then
seek
and
read
so,
as
you
can
imagine
in
the
case
of
a
large
index
file,
this
could
have
long
seek
times
just
for
retrieving
indexes
and
obviously,
in
the
case
of
large
data
files,
this
would
have
a
magnitudes
larger.
Seek
time
now.
A
Remember
when
we
discuss
dynamic
resource
allocation,
where
you
don't
necessarily
need
for
execute
is
a
feeling
of
three
partitions.
Well,
this
scaling
of
your
executors
is
crucial
for
scaling
any
production
spark
environment.
Oh
man
is
critical
at
Bloomberg,
specifically,
so
the
way
that
the
external
shuffle
service
enables
dynamic
of
resource
allocation
is
that,
instead
of
directly
communicating
with
executors
for
the
shuffle
files
via
RPC,
you
could,
when
which
could
be
potentially
scaled-down,
for
example,
in
the
case
of
a
dynamic
environment,
you
would
then
talk
to
a
shuffle
service
instance.
A
Instead,
now
in
a
traditional
spark
on
yarn
environment,
this
shuffle
service
instance
would
live
on
every
node
and
function
as
a
file
server
for
reading
data
files,
as
well
as
a
cache
server
for
index
files
and
as
executors
upon
finishing
its
shuffle
write.
They
would
register
those
written
blocks
with
the
shuffle
service
and
later
the
shuffle
fetches
are
enabled
by
the
shuffle
service
by
streaming
data
from
the
London
nodes,
local
disks
to
the
appropriate
executors.
A
Now
there
are
numerous
problems
with
this
that
we
soon
found,
the
first
of
which
is
that
there's
a
lack
of
resource
isolation.
So
if
the
shuffle
service
uses
too
much
memory
on
the
node,
the
whole
node
goes
down
and
then
takes
on
all
the
executors
that
are
running
on
that
given
node,
there's
also
a
lack
of
replication
in
shovel
service
instances.
A
So
without
multiple
shovel
service
instances
per
node
you'd
have
a
case
where
a
collocated
shuffle
instance,
sales,
for
example,
and
thus
require
replaying
of
the
entire
dag
for
that
for
the
executors
on
that
node,
so
that
the
shuffle
files
can
be
recreated.
However,
they're
all
sending
a
local
disk
anyway.
So,
there's
no
reason
for
you
to
recreate
them.
A
Furthermore,
the
continuous
uptime
of
the
shuffle
service
is
required
for
the
node
to
then
accept
executors
to
run.
So
if
a
shovel
service
goes
down
on
that
node,
then
no
executors
can
be
schedulable
on
that
node
and
lastly,
what
happens?
We
move
this
to
kubernetes,
for
example,
so
in
kubernetes
world
there'd
be
an
assumption
to
container
volume
will
be
shared
between
the
executor
and
the
shuffle
service.
Now
this
design
breaks
in
certain
cluster
configurations
where
co-located
storage
is
not
allowed,
which
I
assume
is
pretty
consistent
across
this
room.
A
So
the
solution
Blumberg
developed
with
Palantir
along
many
developers
from
many
companies
would
be
an
api
within
the
current
shuffle
implementation
that
allows
for
pluggable
writing
and
reading
of
shuffle
bytes.
So
this
API
allows
for
developers
to
extend
their
shuffle
storage
with
any
preferred
backing
storage
layer,
as
well
as
a
database
layer
that
they
can
plug
in
thereby
promoting
disaggregated
compute
for
shuffle
files.
A
Specifically,
the
EPI
leverage
is
a
plugging
tree
that
defines
various
components
and
enabling
shuffle
the
driver
components
allow
for
application,
initialization
and
cleaning
the
executor
components
provide
flexibility
towards
writing
strategy,
may
be
output,
streams
or
read
about
byte
channels
for
Java
neo
optimizations,
like
file
channel
or
sokka
channel,
and
flexibility
for
reading,
as
well
via
the
shuffle
readers
and
the
shuffle
locations.
Allow
developers
the
flexibility
to
create
unique
solutions
that
can
be
comprised
of
primary
and
backup
locations.
So
where
does
it
become
powerful?
A
The
executor
can
then
optionally
handle
replication
by
doing
a
second
rewrite
to
another
file
server,
for
example
the
one
over
here
and
what's
unique
about
that.
One,
for
the
sake
of
example,
is
that,
instead
of
writing,
it's
blocked
to
its
local
file
storage.
It's
writing
to
an
external
file
system
like
HDFS
upon
completion
of
its
writing.
The
executor
will
then
report
below
the
backup
would
then
report
all
of
its
primary
and
backup
in
secondary
locations
to
the
driver.
A
Now
something
to
note
in
this
specific
design
is
that
there
are
two
network
transfers
if
the
file
server
is
communicating
with
an
external
file
system.
The
first
network
transfer
will
then
happen
with
actually
communicating
with
the
file
server
itself,
and
the
second
will
be
communicating
with
HDFS.
A
However,
there's
some
clear
advantages
to
this
design
that
cover
the
problems
that
were
addressed,
every
external
shuffle
service
instance
can
be
run
in
its
own
hosts
or
in
an
or
in
isolated
container.
So
if
any
external
shuffle
service
instance
fails,
only
the
container
running
the
shuffle
servers
will
be
affected.
A
Another
example
on
this
design:
we
are
trying
to
asynchronously
back
up
our
locations
as
we're
writing
our
shuffle
blocks
to
local
disk.
So
upon
writing
our
shuffle
blocks
index
files
to
the
executors
local
file
system,
the
executor
with
an
asynchronously
backup.
It's
shuffle
blocks
index
files
to
a
stable
DFS.
Now
because
it
is
async,
the
executor
can
immediately
report
to
the
driver,
the
primary
and
backup
locations
upon
finishing
it's
local
right.
However,
what
happens
when
we
do
the
read
from
the
backup-
and
it's
not
done
writing,
which
is
quite
common?
A
So
this
solution
is
also
satisfying
for
the
problems
that
we
listed
since
we
removed
the
external
shuffle
service
here,
if
one
executor
fails,
would
only
impact
the
application
that
executors
are
part
of
also
the
backup
storage
layer
would
be
quite
resilient
and
highly
available,
and
the
solution
is
compatible
with
containerized
storage,
since
the
local
disk
does
not
need
to
be
shared
between
any
two
processes
in
the
system.
Exchanging
shuffle
data
would
be
done
either
via
RPC,
between
executors
or
between
RPC,
between
an
executor
and
the
remote
file
servers.
A
Sorry
I'm
in
the
remote
backup
storage,
so
the
last
publication
as
an
example,
is
to
do
a
direct
write
to
distributed
file
system
or
some
kind
of
stable
DFS,
so
where
the
driver
would
maintain
all
of
its
own
shuffle
metadata
without
using
a
shuffle
service,
and
the
executor
can
then
circumvent
the
need
for
a
file
system
and
would
directly
write
its
shuffle
blocks
index
files
to
the
stable
of
DFS.
So
something
to
note
here,
in
contrast,
the
individual
file
server
example
that
was
earlier.
A
You
now
only
require
one
one
network
transfer,
as
you
directly
communicate
with
the
DIA
and
for
further
optimizations.
You
could
also
leverage
an
in-memory
key
value
store
like
ignite
or
Redis,
and
you
actually
remove
the
need
for
index
file
completely
by
just
having
the
partition
to
be
part
of
that
key.
A
This
example
also
solves
the
problems
they
were
addressing
earlier
because
it
provides
the
necessary
resource.
Isolation
also
with
the
scalability
factor
that
the
scalability
factor
only
relies
on
the
scalability
of
the
shuffle
operation,
as
well
as
the
scalability
of
the
the
data
storage
system,
which,
in
this
case
is
going
to
meet
quite
performant.
A
So
upon
completion
of
a
proposed
API
and
despite
making
some
headway
on
trying
to
run
some
of
our
SPARC
deployments,
you
use
these
implementations,
there's
still
a
lot
of
work.
That
needs
to
be
done.
Obviously
so
the
API
and
some
of
the
implementations
I
present
here
are
formalized
in
an
SPI
P
under
the
JIRA
tickets,
SPARC
two
five,
two
nine
nine
and
is
an
active
discussion
of
the
other
spark'd
of
mailing
lists.
A
However,
in
the
work
of
enabling
a
fully
disaggregated
SPARC
environment,
there
are
still
challenges
that
remain
so
the
work
the
slim
needs
to
be
done
involves
fetch
failed
exception
handling
and
the
dag
scheduler
for
failed
shovel
fetches,
which
is
quite
common.
So
these
shall
fetch
failures,
our
primary
sources
of
query
latency
in
SPARC
and
as
such,
we
will
like
the
we
would
like
the
implementer
to
customize
the
exception
that
is
thrown
for
their
given
specific
implementations.
A
So,
for
example,
in
the
case
of
your
direct
writing
to
DFS
your
have
situation
where
you
don't
want
necessarily
do
a
fetch
failure,
because
you
want
to
do
generic
exception,
because
the
chances
of
you
being
unable
to
freshen
DFS
is
quite
low
because
of
the
scalability
of
your
vacuum
store.
The
next
thing
we
have
to
do
is
extend
the
API
to
handle
spills
and
cache
files,
so
cache
files
is
another
kind
of
temporary
file
that
I
didn't
get
into
in
this
talk.
Luckily,
these
two
api's
are
quite
orthogonal
to
our
work,
and
so
we
don't.
A
We
still
need
to
necessarily
extend
them
to
make
this
fully
supported,
but
luckily
we
also
pretty
much.
Those
are
the
last
two
things
that
are
needed
to
fully
move
that
cash
flow.
Caching
layer
outside
of
our
compute
nodes
didn't
fully
promote
to
sag
and
the
last
things
to
do
performance
benchmarking
across
different
implementations
before
deciding
which
one
is
appropriate,
Sparks
environment.
As
always,
we
encourage
collaboration
as
we
move
towards
disaggregated,
compute
I'm.
A
B
A
He
wanted
to
do
local
file
reads,
and
so
a
lot
of
that
was
in
pretty
much
you're
using
too
many
kubernetes
specific
things
in
an
environment.
That's
supposed
to
be
agnostic
to
cluster
managers.
So
there
was,
there
was
just
back
and
forth
between
the
community
where
we
needed
to
redesign
it
and,
as
I
discussed
earlier,
the
new
design
of
the
generic
API
promotes
the
ability
for
us
to
have
various
different
shuffle
mutations.
Just
leverage
these
wrappers
allow
for
you
to
directly
call
write
output
stream,
read
output
stream
and
various
shuffle
locations.
C
A
So
you
pretty
much
define
a
resource
quota
or
your
name
stays,
so
our
workspaces
are
defined
around
a
namespace
and
then
based
not
use
calculate
the
size
of
your
cluster.
That
you're
specifying
like.
Let's
say
when
you
launch
a
spark
job,
you
request
executors
memory
and
Driver
memory
just
based
on
that
calculation.
You
would
then
determine
if
this
job
is
too
large
and
the
question
of
how
you
would
handle
scheduling
the
gang
scheduling
or
FIFO
or
whatever
of
that
nature
between
spark
jobs.
A
That's
still
under
investigation,
whether
there's
an
appropriate
place
for
that
in
terms
of
how
you
schedule
jobs
in
terms
of
reach
or
scheduling.
But
the
simplest
approach
would
then
be
just
like
whether
you're
over
prevent
your
pretty
much
you're,
giving
too
many
reasons
to
your
spark
jobs
and
be
blocked
from
launching,
if
that
makes
sense,
I.
D
Hello,
we
are
currently
running
spark
on
young,
but
most
the
reason
is
because
we
have
a
pool
of
resource
which
is
purely
just
the
full
spark,
and
we
have
many
researchers
running
in
Sparks,
your
opponent's
for
young.
They
have
this
ability
to
preempt
jobs.
So
when
the
cluster
is
empty,
we
allow
one
researcher
to
just
fully
use
the
whole
cluster,
but
when
the
second
researchers
join,
singing
young
will
make
sure
it
split
this
cluster
half
and
a
half
to
make
it
fair
for
the
two
researchers,
but
on
Cuban
artists.
D
A
A
A
E
My
question
is
about
performance.
Currently,
the
shuffle
service
is
is,
is
a
file
server
that
tries
to
the
local
file
system
and
it's
one
of
the
of
the
biggest
issue
for
spark
graphs
perform
performance.
So
when
you
move
to
writing
directly
to
HDFS
Torn
external
HDFS
cluster,
that
does
it
have
you
use
it
on
big
workloads
on
big
spark
drops
so.
A
As
the
problem
question,
I
guess
comes
down
to
like
query
latency
in
general,
like
when
you
have.
How
long
does
it
usually
take
most
of
the
time
that's
spent
on
shuffle
and
so
becoming
an
issue
with
regards
to
fetch
failed
exceptions,
so
your
fetch
shell
exception
is
going
to
cause
a
predominantly
large
portion
of
variability
in
your
spark
job
times
and
so
to
stabilize,
at
least
that
by
ensuring
they're
writing
to
a
stable
DFS
that
should
at
least
provide
users
consistency
in
their
spark
jobs,
as
well
as
the
ability
for
it
to
be
faster.
A
We
also
see
that
network
and
network
itself
has
becoming
more
and
more
developed
in
general,
so
we're
saying
that
our
network
costs
are
gonna,
go
down.
Well,
our
determining
our
disk
throughputs
we're
seeing
that
we're
not
gonna.
We
were
focusing
more
on
network
than
disk
throughput
and
so
we're
seeing
that
when
he
comes
to
larger
jobs,
we
want
to
focus
more
on
not
relying
on
disk
throughput,
as
that's
hot
you'd,
say
the
industry's
not
moving
more
towards
that
as
opposed
to
network.
A
The
other
thing
is
that
there's
some
jobs
that
physically
you
can't
so
there
were
some
other
talks
by
Facebook
during
there
and
during
spark
summit
where
they
were
discussing,
write
amplification.
Do
you
just
shuffles,
and
so
in
some
cases
you're
for
example,
one
petabyte
of
shuffle
operation
can
results
in
three
times
that
so
three
petabytes
of
shuffle
files
in
spell
files
written
right,
so
that
physically
the
the
ability
to
write
that
all
in
disk
is
sometimes
impossible,
and
so
there
is
a
need
then,
to
write
to
DFS.