►
From YouTube: Managing Multi-Cloud Apache Spark on Kubernetes - Ilan Filonenko, Aki Sukegawa, Bloomberg
Description
Managing Multi-Cloud Apache Spark on Kubernetes - Ilan Filonenko, Aki Sukegawa, Bloomberg
Bloomberg has built multi-cloud quant platforms on top of Kubernetes to enable its users to develop sophisticated financial applications with integrated first-class data science capabilities. In this journey, it quickly became clear that managing data science infrastructure in a multi-cloud environment is challenging, especially when it comes to Apache Spark. While Kubernetes provides an excellent abstraction for designing composable infrastructure substrates, it comes with a list of challenges when dealing with auto-scaling, scheduling, preemption, and security. Given these challenges, this talk will explore how one can effectively manage an expansive Spark infrastructure solution that spans bare-metal and multiple public cloud platforms. We will also walk through various observability strategies, primarily focusing on how to surface cluster information to a varied group of Spark end-users by leveraging a variety of native Kubernetes resources, like node autoscalers, controllers, and custom PodConditions.
A
Okay
testing,
we
good
okay,
awesome,
okay,
welcome
everyone,
and
thank
you
for
coming.
My
name
is
alan
felinenko
and
I'm
an
engineering
team
lead
at
bloomberg,
and
I'm
joined
here
by
my
colleague,
aki
sukagawa
who's,
a
senior
software
engineer
with
me
on
the
enterprise
data
science
infrastructure
team.
So
today
we'll
be
talking
about
managing
multi-cloud
apache
spark
on
kubernetes,
so
in
general,
starting
from
the
top
managing
data
science
infrastructure
is
in
a
multi-cloud.
Environment
is
really
hard.
The
variation
of
behavior
between
different
hardware
and
cloud
providers
is
vast,
and
rather
daunting.
A
A
So
what
is
apache
spark
show
of
hands
who's
familiar
here
with
spark
and
who's,
not
good?
Okay,
so
there's
a
couple:
okay,
there's
a
couple,
so
we'll
do
a
little
bit
of
a
rundown.
So
what
is
apache
spark?
So
the
quick
background
for
those
who
are
unfamiliar
is
that
it's
an
etl
application
that
enables
us
to
provide
data
science
at
scale
so
traditionally
etl
for
files
smaller
than
the
gigabyte
uses
pandas,
but
we
start
scaling
up
to
100
gigabytes.
A
We
rather
need
to
chunk
pandas
data
frames
or
use
das,
but
spark
can
also
be
a
pretty
good
candidate,
but
where
it
really
shines
is
when
you
start
looking
at
petabytes
or
more,
it
has
a
variety
of
features
that
make
it
a
powerful
general
execution
engine,
especially
when
communicating
with
cloud
cloud
object
stores,
but
for
the
purpose
of
this
talk
we'll
be
focusing
on
its
distributed.
Architecture
and
its
relation
to
kubernetes
spark
runs
as
independent
processes
on
a
cluster
coordinated
by
the
spark
context.
A
Object
in
your
driver
program,
so
spark
context,
can
connect
to
several
types
of
cluster
managers
like
mesos,
yarn
or
kubernetes,
for
the
and
for
the
purp
for
the
purpose
of
allocating
resources
across
applications
so
sparks
cluster
manager.
Functionally
is
just
purpose
to
provide
ip
addresses
with
which
spark
uses
them
to
communicate.
So
this
enables
the
cluster
managers
to
be
rather
flexible
or
composable
for
a
variety
of
different
scheduler
back-ends
like
yarn
or
kubernetes.
A
Now
this
has
benefits
for
plugability
via
scheduler
interface,
but
this
does
mean
that
spark
isn't
good
at
looking
up
or
trying
to
bubble
up
too
much
kubernetes
information,
as
it
won't
necessarily
fit
into
the
general
interface.
We
call
this
kind
of
a
lack
of
information
flow,
and
this
is
a
lot
of
drawbacks
that
we'll
talk
about
throughout
this
talk
so
upon
receiving
the
provision.
Executors
spark
sends
the
application
code
to
the
executors,
after
which,
then
the
spark
context
would
send
tasks
to
the
executors
to
run.
A
So
if
we
were
to
parallel
the
kubernetes
speak,
we
could
kind
of
think
of
of
the
driver
here
as
a
controller
of
sorts.
It's
now
tasked
with
provisioning
and
managing
state.
However,
this
is
all
done
from
within
spark
are
folks
familiar
with
controllers
or
kubernetes
controllers
sweet
now.
What
does
it
mean
for
for
a
driver
to
have
controller-like
privileges?
So
this
means
that
the
user's
pod
will
have
elevated
service
account
privileges
that,
in
some
environments
like
ours
at
bloomberg,
is
rejected
due
to
security
concerns.
A
Another
potential
problem
is
that
the
full
information
flow
from
kubernetes
is
expected
to
reach
spark,
but
because
of
the
agnostic
nature
of
the
schedule,
as
I
said
earlier,
this
might
not
be
entirely
possible
and
we'll
and,
as
we
discussed
a
little
bit
earlier
so
to
solve
this
bloomberg.
As
I
also
talked
about
my
previous
coupon
talk,
looked
to
enable
a
plugable
executor
creation
template
that
pushed
the
pod
creation
to
an
admin
controller.
A
So
this
allowed
now
for
the
communication,
with
the
admin
controller,
to
kind
of
push
out
that
surface
account
requirement
now
to
communicate
with
controls
in
general.
This
is
traditionally
done
via
custom
resources,
so
this
means
that
the
spark
driver
now
modifies
a
custom
resource
instead
of
creating
pods
directly.
A
So
to
help
understand
this,
let's
work
with
the
visual
your
driver
pod,
which
in
this
example,
is
running
within
the
user's
namespace
from
a
jupyter
note
from
a
jupyter
notebook
pod,
would
communicate
with
the
api
server
to
create
and
modify
a
spark
app
custom
resource
with
the
intention
of
having
the
controller
and
the
admin
namespace
create
those
executors
securely
in
the
user's
namespace.
The
admin
controller
will
watch
for
spark
app
custom
resources
where
the
spec
field
of
these
custom
resources
has
some
desired
state.
A
In
our
case,
we're
specifying
the
expected
executor,
pod
templates
and
the
controller
will
then
act
on
the
contents
of
that
spec
field
by
creating
these
respective
executor
pods
via
its
reconciliation
loop.
The
drive
will
then
wait
for
those
executor,
pods
asynchronously,
which
is
functionally
done
the
same
as
it
would
otherwise.
So,
therefore,
the
rest
of
the
spark
paradigm
kind
of
remains
the
same.
Even
with
this
plugable
design.
A
Now
we've
been
running
our
personal
fork
of
spark
in
a
managed
data
science
platform
across
bare
metal
and
various
clouds
for
some
time.
So,
let's
walk
through
a
couple
user
studies
that
address
the
success
stories
and
complications
that
arose.
So,
let's
start
with
all
of
our
success
stories,
we
got
a
comment
that
it
works.
A
Okay,
now
for
what
matters,
the
complications
that
arose
so
without
spending
too
much
time
on
each
of
these
issues,
as
you
can
review
them
in
their
solutions
offline.
I
think
we
posted
the
slides
what
I
want
to
highlight
here.
There
are
two
categories
of
issues:
ones
relating
to
the
pending
status
of
the
executors
and
the
others
relating
to
the
failed
states.
A
So
the
three
generalized
limitations
that
result
from
missing
information
flow
include:
auto
scaling
information,
the
capture
of
scale,
down
preemption,
om
and
failure,
events
and
post
job
introspection
where
a
user
might
want
to
triage
their
spark
job
after
the
job
completes
or
fails.
So
this
complements
their
existing
monitoring
or
logging
dashboards
that
might
be
available
in
their
platform,
so
I'll
now
pass
it
over
to
aki
so
that
he
can
walk
us
through
our
approach
and
our
solutions
to
addressing
these
limitations.
In
apache
spark.
B
So
our
solution
to
the
lack
of
information
is
to
collect
information
ourselves
somehow
and
store
high-level
statistics
into
sales
object.
I
mean
status
field
of
the
custom
resource
we
described
earlier.
So,
as
you
can
see
in
this
slide
missing
all
scaling
information
is
now
stored.
I
mean,
will
be
stored
in
the
cluster
scaling
field
on
the
right
hand,
side
and
scaled
down
and
oom
out
of
memory
failures.
Information
are
stored
in
a
terminal
status
field
and
first
for
process
of
introspection.
B
B
We
can,
for
example,
build
a
ui
on
top
of
the
information
we
gather
from
the
custom
resource
or
bubble
up
the
information
to
the
jubilee
notebook
that
is
running
the
driver
program,
so
we
could
directly
collect
the
information
and
provide
an
endpoint
to
get
the
statistics.
Statistics
in
one
service
there's
nothing
wrong
with
this
approach,
but
we
prefer
having
a
separate
component
for
just
collecting
the
information
inside
the
kubernetes
world.
B
So
by
having
a
controller
to
collect
information
from
underlying
vanillis
layer,
we
have
a
nice
separation
of
concern
and
also
we
can
follow
a
common
controller
pattern
for
handling
kanellis
objects.
So
what
does
it
mean?
So,
as
we
saw
earlier,
we
use
the
spark
up
custom
resource
for
creating
executables.
B
So
we
are
following
the
control
pattern
here
when
we
store
the
statistics
about
the
executors
to
the
status
field
back
so
other
considerations
about
the
controller
pattern
itself
is
that
its
reconciliation
load
needs
to
be
it
important,
meaning
that
it
needs
to
be
stateless.
It
can't
rely
on
the
previous
results,
so
it
works
from
scratch
every
time
the
reconciliation
loop
is
invoked.
B
So
another
way
to
think
about
this
approach
is
that
we
are
effectively
using
the
custom
resource
as
a
data
data
store
for
the
high
level
statistics
we
want.
We
are
effectively
storing
information
to
scd
via
kubernetes
api
server.
So
here's
the
characteristics
of
this
approach
as
a
data
store,
so
it
it
introduces
no
extra
dependency
and
it
provides
out
of
the
box,
publish
up
a
mechanism
via
watch
and
we
can
do
basic
query
and
aggregation
by
selectors
and
controllers
and
the
latency
is
not
ideal
but
isn't
too
high
as
well.
B
B
The
latency
requirement
for
our
ui
is
not
so
tight
and
we
only
need
the
basic
query
mechanism
in
this
case
in
our
use
case,
and
since
we
run
it
in
bare
metal
and
multiple
cloud
cloud
services,
we
really
want
to
minimize
extra
dependencies
in
our
case,
because
the
maintenance
cost
and
the
complexity
in
each
environment
add
up
so
there
are.
There
are
two
alternatives
we
considered
as
a
data
source.
B
So
another
way
we
thought
about
is
to
use
an
additional
database.
This
is
suddenly
doable,
but
it
introduces
extra
infrastructure
and
we
also
need
to
choose
our
solution
carefully
and
design
on
top
of
it
with
care
as
well.
So,
as
we
saw,
we
are
effectively
choosing
scd
for
our
database
for
the
advantages
listed
before
now
that
we
have
proposed
using
a
custom
resource
to
store,
execute
as
high
level
statistics.
B
Let's
work
through
each
of
the
limitations
in
information
flow
we
identified
earlier,
we
will
start
with
auto
scaling.
So
what
is
all
the
scaling
up?
So
cluster
scale
up
happens
when
existing
node
existing
kubernetes
nodes
cannot
fit
in
newly
created
ports.
So
if
there's
not
enough
nodes
already,
the
cluster
scalar
tries
to
create
more
node.
So
when
it
happens,
there
are
actually
two
problems
from
the
perspective
of
use,
spark
user
experience.
B
It
creates
an
event
object
with
a
reference
to
the
affected
pod,
like
shown
in
the
slide,
and
they
create
so
the
the
references
in
the
involved
object
field.
So
one
reason
we
chose
to
run
the
standard
cluster
scalar
by
the
way
was
because
of
our
multi-cloud
environment
and
the
desire
to
keep
a
single
implement
implementation
across
all
deployments.
B
So
this
is
another
example
for
when
a
cluster
scalar
app
was
impossible,
so
the
last
one
was
for
scaling
up
was
possible
and
triggered,
and
we
can
so.
We
can
detect
cluster
scaling
state
of
each
executable
by
letting
these
events
from
cluster
autoscaler
involving
the
particular
port
and
pick
the
latest
one.
B
B
So
how
we
do
this
so
one
way
is
to
put
the
custom
label
or
annotation
to
the
pod.
We
define
an
arbitrary
custom
level
name
and
the
projector
information,
with
some
constraints
and
another
less
known
way,
is
to
put
the
custom
condition
to
the
pub
status
field.
We
can
define
a
custom
condition
type
with
additional
field
with
more
flexible
constraint
now
that
we
handle
the
scale
up.
The
next
problem
that
we
will
work
on
is
a
cluster
of
scaling
down
and
out
of
memory.
B
B
B
So
one
way
to
deal
with
this
deletion
is
not
to
delete
pods
in
the
first
place,
so
there
is
a
relevant
option
introduced
in
spark
30
that
prevents
spark
drivers
from
the
deleting
failed
executive
ports
by
default.
Spark
driver
deletes
failed
bots,
but
by
setting
this
deleted
deletion
termination
option
to
false
it
leaves
the
failed
pause
as
they
are.
B
So
this
solved
the
majority
of
the
problem
and
it
dissolved
the
issue
for
out
of
memory
era,
but
this
only
prevents
spark
driver
from
killing
pulse
and
there
are
others
who
can
kill,
exhibit
catapults,
for
example,
in
the
case
of
a
cluster
auto
scale
down
or
eviction,
port
will
be
deleted
by
kubernetes
core
components,
not
drivers.
So
it's
not
prevented
by
the
above
config.
B
It
turns
out
that
even
for
those
cases,
we
can
actually
prevent
the
deletion
by
using
kubernetes
finalizers
finalizes
is
a
field
in
the
metadata
and
it's
just
an
array
of
strings
and
as
long
as
it's
not
empty,
the
object
cannot
be
deleted,
so
we
can
put
a
custom
arbitrary,
finalizer
there
and
prevent
execute
the
part
from
being
deleted.
As
long
as
we
want.
B
We
did
not
choose
this
approach,
because
the
scope
of
the
implication
is
really
hard
to
know
how
it's
such
a
central
component
of
kubernetes,
and
so
many
types
of
resources
are
attached
to
it
that
it
it's
very
hard
to
foresee.
Every
single
edge
cases
we
can
face
down
the
road,
so
we
decided
that
we
cannot
keep
the
pod
resource
itself.
We
need
to
store
the
information
somewhere.
B
B
So
this
is
what
pod
stylus
custom
resource
actually
looks
like.
We
follow
the
common
pattern
of
spec
and
status
field
here
and
spec
is
a
desired
state
and
the
status
is
the
realized
state
and
in
this
case
it's
kind
of
a
little
bit
very
it's
a
variant
of
that
spec
field
specifies
the
target
part
to
track
in
the
object
and
the
controller
replicates
the
necessary
information
into
the
status
field,
meaning
that
the
status
field
itself
is
the
realized
state.
B
So
we
can
selectively
activate
this
functionality
for
a
single
port
by
creating
one
public
service
object
like
this
so
optionally.
We
can
have
a
separate,
webhook
or
controller
on
top
of
this,
and
they
automatically
create
such
power
source
objects
for
each
part
that
matches
certain
criteria
like
in
spark.
There
are.
There
is
a
requirement
that
all
executor,
paws
have
labeled
spark
role,
equals
executable.
B
B
So,
in
the
for
this
controller
we
have
a
configuration
like
this
and
in
the
left
hand,
side
we
specify
the
source
and
destination
field
degradatively
and
the
type
of
course,
and
the
controller
will
watch
the
source
objects
and
keep
replicating
specified
fields
resulting
destination.
Object
in
this
case
is
shown
in
the
right
hand,
side
and
the
owner
reference
field,
and
the
container
status
field
is
replicated,
and
this
is
another
example
for
the
cluster
scalar
event
we
on
top
on
top
of
the
previous
one.
B
A
downside
of
this
approach
is
that
everything
becomes
a
goal
map
in
the
implementation
and
we
lose
static
typing
you
might
want
to
do
more
than
just
copying,
for
example,
define
more
complicated
trigger
rule
and
have
more
versatile
actions
triggered
by
those
rules,
not
just
copying,
etc,
and
it's
totally
possible
and
not
that
hard
to
do.
But
we
would
recommend
keep
the
configuration
like
this
really
simple,
as
you
don't
really
want
to.
For
example,
you
need
to
test
yam
or
configuration
down
the
road.
B
B
So,
in
other
words,
you
can
make
sure
that
all
the
business
logic
resides
in
the
main
controller
and
not
randomly
spread
across
multiple
places,
but
in
our
use
cases
we
actually
wanted
to
encapsulate
autoscaler
related
logic
into
additional
controller,
the
polar
cells
controller.
So
we
didn't
choose
this
route.
B
B
D
A
Got
it
so
it's
general
data
management.
Do
you
guys
hear
me
sorry,
general
data
management,
questions
yeah.
The
context
of
the
talk
is
mostly
about
the
compute,
but
in
terms
of
data
it
varies
between
like
the
restrictions
and
the
security
concerns.
The
application
like
if
it's
something
where
we're
taking
data,
that's
only
within
a
spurt
in
like
say,
network
work,
we're
constricting
it
to
would
have
like
calculated
policies,
locking
everything
down
in
terms
of
moving
data
in
and
out
right,
and
so
it
varies
based
on
the
implementation.
A
So
when
you
see,
when
you
say
the
okay,
the
pods
themselves
are
on
the
user's
name,
space
right,
and
so
the
data
itself
is
either
collocates
in
the
name
space
or
you
have
specific
verification
on
what
data
source
you're
communicating
to
so
the
pods
themselves
are
the
user
namespace.
The
admin
controllers
are
just
a
control
plane
right.
The
paws
are
not
in
the
control
plane,
it's
only
in
the
user's
name
space
for
the
spark
app.
Does
that
answer
the
question.
C
Okay,
so
I
yeah
there
was
one
behind
hey.
I
have
two
questions.
The
first
one
is:
you
had
on-prem
data
center
footprint.
You
had
aws
and
gcps.
I
was
curious
if
all
the
cluster
compute
nodes
on
all
these
three
footprints
across
cloud
providers
are,
are
they
homogeneous
or
do
you
have
heterogeneous
computer
environments.
B
A
Yeah,
so
the
we
were
using
the
standard
cluster,
auto
scaler
from
kubernetes
right.
The
intention,
as
aki
mentioned,
was
that
because
we're
in
a
multi-cloud
environment-
and
we
need
portable
infrastructure,
the
built-in
standard
cluster,
auto
scale
or
something
that
would
run
in
eks
aks
and
on
bare
metal,
because
it's
all
kubernetes
abstraction
right.
So
we
just
use
the
base
cluster
autoscaler,
so
we
didn't
build
our
own,
but
the
idea
is
because
we
have
a
separation
of
concerns
between
the
job
controllers
that
reconcile,
however,
that
job
operates
and
the
auto
scaling
information.
E
A
So
with
regards
to
kind
of
considerations,
so
traditionally
it's
kind
of
about
like
what
is
your
spark
app
clusters
like
scale
to
right,
like
we
had
some
battle
tests,
I
think
400
and
500
kind
of
executor
clusters
and
that's
kind
of
what
I've
seen
mostly
like
sub
1000.
I
don't
really
know
too
many
have
more
than
a
thousand
kind
of
executor,
pods.