Manipulating big data distributed over a cluster using functional concepts is rampant in industry, and is arguably one of the first widespread industrial uses of functional ideas. This is evidenced by the popularity of MapReduce and Hadoop, and most recently Apache Spark, a fast, in-memory distributed collections framework written in Scala. In this course, we'll see how the data parallel paradigm can be extended to the distributed case, using Spark throughout. We'll cover Spark's programming model in detail, being careful to understand how and when it differs from familiar programming models, like shared-memory parallel collections or sequential Scala collections. Through hands-on examples in Spark and Scala, we'll learn when important issues related to distribution like latency and network communication should be considered and how they can be addressed effectively for improved performance. Learning Outcomes. By the end of this course you will be able to: - read data from persistent storage and load it into Apache Spark, - manipulate data with Spark and Scala, - express algorithms for data analysis in a functional style, - recognize how to avoid shuffles and recomputation in Spark, Recommended background: You should have at least one year programming experience. Proficiency with Java or C# is ideal, but experience with other languages such as C/C++, Python, Javascript or Ruby is also sufficient. You should have some familiarity using the command line. This course is intended to be taken after Parallel Programming: https://www.coursera.org/learn/parprog1.
Wide vs Narrow Dependencies
Loading...
来自 École Polytechnique Fédérale de Lausanne 的课程
Big Data Analysis with Scala and Spark
1529 个评分
École Polytechnique Fédérale de Lausanne
1529 个评分
课程 4（共 5 门，Specialization Functional Programming in Scala）
从本节课中
Partitioning and Shuffling
This week we'll look at some of the performance implications of using operations like joins. Is it possible to get the same result without having to pay for the overhead of moving data over the network? We'll answer this question by delving into how we can partition our data to achieve better data locality, in turn optimizing some of our Spark jobs.
与讲师见面
Dr. Heather MillerResearch Scientist
EPFL
In this session,we're going to focus on wide versus narrow dependencies,
which dictate relationships between RDDs in graphs of computation,
which we'll see has a lot to do with shuffling.
A recurring theme lately has been that, not all transformations are exactly equal.
Some are a lot more expensive in terms of latency than others.
For example, some might require lots of data to be transferred over the network
perhaps unnecessarily so, operations that cause a shuffle are an example of this.
In order to better understand when shuffling might occur, we have to look at
how RDDs are represented, and then we have to think carefully about the operations
that we're calling on our RDDs, to understand when a shuffle might occur.
We'll actually see that this unique way of representing RDDs,
actually makes fault tolerance possible in Spark.
So we mentioned in earlier sessions that Spark is fault tolerant,
we'll get a glimpse of how that's possible and actually kind of easy.
So let's start with some terminology.
Let's start with lineages.
So if you think about a group of computations that are done on a RDD,
we can call that group of computations, a lineage graph.
That is when you do operations on RDD, the operations can be organized into
a Directed Acyclic Graph, representing the computations that were done on that RDD.
So here's an example,
we start with some input RDD, then we call the number of operations on that RDD.
We can then reuse one of these RDD's in two subsequent computations.
So we can reuse this filtered RDD in account, and
we can also reuse it in a reduce.
If you try to visualize it, it makes a graph like this.
An rdd is the rdd that's actually made from this input file.
We can then do map and filter, these transformations here are not rdd.
And we'll get back in new RDD called filter, which we could persist in memory.
And then we can use that filtered RDD in two different ways, with a count and
with a reduce operation.
This forms a graph of computations of tree without cycles.
That's why it's called a Directed Acyclic Graph.
In Spark, these graphs or DAGs are referred to as the lineage graph.
That means it's possible to step back up this graph,
to figure out how the result of this count operation is derived from input file.
These lineage graphs are a big part of how RDDs are represented in Spark.
And in fact, Spark can actually analyze this representation and do optimizations.
We'll see what that means in a moment.
But first, let's look at how RDDs are represented.
So let's start with a visual example.
Let's assume that this block here is an RDD.
Usually, we think of RDDs as some kind of abstraction that we invoke methods on.
But an RDD itself is actually made up of Partitions, partitions are atomic pieces
of the dataset, and they may exist on one or many compute node.
So you might have one RDD spread out over many compute nodes, and
all of the pieces that are spread out are the Partitions.
The next most important part of an RDD are Dependencies.
Dependencies model relationship between an RDDs partitions,
and the partitions that it was derived from in other RDDs.
So for example in this picture, if this is a parent RDD and
this is a child RDD, and we're calling the map function on this RDD to get that RDD.
Then the Dependencies are how these partitions mapped to those partitions with
the map function.
So, while you might normally think of a map as one big arrow between this RDD and
that RDD, actually Dependencies model the relationship between some partitions and
the partitions that they were derived from.
So you could say in this picture that this partition was derived from that partition.
This Partition was derived from that partition and so on.
And finally the last two parts of an RDD include a function, so
you can think of this as the function that you might you might pass to
a map operation on the previous slide, you can't compute a child partition without
some kind of function to get that child partition from the parent partition.
So functions are actually a big part of RDDs,
because they say how to compute the data sets based on an RDDs parents.
So what to actually do.
And finally, there is some metadata about the partitioning scheme and
where data is placed that's a part of an RDD.
So these are the four parts of an RDD.
But you might ask well how does that relate to shuffles?
If you recall in a previous session, we developed the following rule of thumb.
We said a shuffle can occur,
when the resulting RDD depends on other elements from the same RDD or another RDD.
So here's a key word, depends.
Now we have an idea of what a dependency is, in
fact an RDD's dependencies actually encode when data must move over the network.
So that brings us to transformations.
So we know that transformations cause shuffles, and
we know that dependency information can tell us when a shuffle might occur.
To differentiate between these, we can actually define two sets of dependencies
called Narrow and Wide Dependencies, that can tell us when shuffles will happen.
We're going to go more into depth about what a Narrow Dependency and
a Wide Dependency means in this session.
So let's define Marrow and Wide Dependencies.
A transformation has Narrow Dependencies, when each partition of
the parent RDD is used by at most one partition of the child RDD.
So that means some child partition has only one parent partition and not many.
A transformation that has Wide Dependencies on the other hand,
is one where each partition of the parent RDD
may be depended on by multiple children partitions.
So that means you may have many child partitions,
which were all derived from a single parent partition.
Transformations with these kind of dependencies have Wide Dependencies.
So what does that mean to us?
Well, as you might've guessed,
transformations with Narrow Dependencies are typically quite fast.
They require no shuffles, and optimizations like pipelining can occur.
Which is when you can group together many transformations into one pass.
And as you may have guessed, transformations with Wide Dependencies
are slow, because they require all or some data to be shuffled over the network.
So these are the transformations that cause shuffles.
So let's try to visualize the differences between Narrow and Wide Dependencies.
Let's start first with Narrow Dependencies.
So as I said on the previous slide,
each partition of the parent RDD is used by at most one partition of the child RDD.
So some example transformations include map and
filter, because one partition depends on at most one other partition.
And the same is true for union.
We can essentially just put the data in this partition,
into the new resulting RDD here.
There's no relationship between many RDDs,
when it comes to looking at who this partition was derived from.
And what might surprise you, is that this is also true for
joins that are already partitioned.
So if we recall in previous sessions,
how we actually were able to create joins where shuffles weren't required,
this is the sort of join which has a Narrow Dependency.
But just to go back to the definition of a Narrow Dependency,
join with co-partitioned inputs, is a transformation with narrow dependencies.
Because each partition of the parent RDD, is used but
at most one partition of the child RDD.
That is this child dependency here,
is derived from only one partition in one of the parent RDDs.
So in another way,
we don't have an error going from this partition to that child partition, and
then also this partition here going to that same first child partition.
We have just one relationship, this partition relates to that partition and
that's it.
On the other hand,
transformation with Wide Dependencies as we said on the previous slide, is when
each partition of a parent RDD may be depended on by multiple child partitions.
So one operation that we already know caused the shuffles is groupByKey.
Now when we visualize groupByKey with partitions and dependencies,
we can see how a shuffle will occur.
As the definition here says,
we have parent partitions which are depended on by multiple child partitions.
So for example, this partition is depended on by both of the children partitions.
So these are Wide Dependencies.
Another operation with Wide Dependencies of course is join,
when it's inputs are not already partitioned, this means that we are going
to have to move data all over the network to make sure that the resulting RDD
has all the values corresponding to some key on the same partitions.
Before the join is called, they could be spread all over the network.
So this is a pretty clear example of an operation that has Wide Dependencies,
a join without already partitioned inputs.
So let's take this visualization a step further, and
let's visualize a sample program and its dependencies.
So let's assume we have the following DAG, the following graph of computations.
Where each one of these here are RDDs.
So we have this box is an RDD, this one's called A, this one's called B,
this one's called C, D, E, F, G.
They each have different numbers of partitions.
So A has three partitions, C has two partitions.
And these big arrows here,
represent the operations that we're calling on those RDDs.
So A is the parent RDD.
And B is the child RDD, and the operation that B is derived from is groupBy.
So here's a quick exercise for you.
If you had to draw the dependencies between each of these RDDs,
what would they look like?
Which dependencies are wide and which dependencies are narrow?
Try it for yourself.
Try to draw it out on a piece of paper, and think very carefully about what
functions like groupBy and map, actually semantically do.
Ask yourself how many dependencies each one of these partitions might have on
their parents.
Well, here's the answer.
If you think very carefully about what groupBy has to do,
it has to group values by the same key on in the resulting RDDs.
So it has to move key value pairs around the network,
in order to make sure that they're all in the same partition.
Which means that many of the parent RDDs partitions,
are used in each partition of the child RDD.
Operations like map have only one dependency.
So this partition depends on only that partition and
it doesn't matter what's in this partition.
So the next question I asked you, are which dependencies are wide and
which are narrow?
Transformations like groupBy and join are wide,
because each child partition depends on multiple parent partitions.
And transformations like map and union are narrow,
because each child depends on only one parent RDD.
There's a one-to-one relationship between parent partitions and child partitions.
Well, hang on a second. You might be asking yourself, well,
why is this side of the join narrow?
And the simple answer is that if you looked at the code,
you probably cached B in memory because you knew you were doing a group by key,
which does partitioning already.
So you want to keep that partitioning in place when you reuse B.
So I said simply, B then is already co-partitioned.
That's already partitioning in cached memory following the call to this
groupBy here.
That's why this part of the join can have Narrow Dependencies.
So I hope those visualizations were an aha moment for you,
when thinking about computations that you have to do in Spark.
You actually have to think carefully about whether or
not an operation that you want to use, might have Narrow or Wide Dependencies.
And of course we saw that sometimes some operations like join,
sometimes are narrow and sometimes are wide.
So, I can make a list here of operations that are usually always narrow and
usually always wide.
Though there of course some exemptions such us join here,
because we know that join is not always wide.
Sometimes it can be narrow depending on the partitioning scheme.
But here we can roughly breakdown transformations into narrow and wide.
Again, doesn't always hold but it's a rough approximation.
So you can always look back at this list and use it as little bit of our reference,
but it's not always going to give you the answer.
You're still going to have to think carefully about the operation that you're
actually doing on the cluster.
But sometimes, maybe you want to be a little bit more certain than
thinking critically about the operations that you're using.
For that, Sparks gives you a method called dependencies.
And dependencies will give you a sequence of dependency objects,
which are actually the dependencies that Spark uses in its scheduler,
to know how this RDD depends on other RDDs.
When you call this method,
you'll get back a sequence of dependency objects with different names.
Dependencies called OneToOneDependencies are Narrow Dependencies.
You might also see something called a Prune and a Range Dependency.
But most often you'll see OneToOneDependency for
narrow transformations.
And finally wide dependencies are called ShuffleDepedency in Spark,
which is a pretty clear indicator that a shuffle might occur.
Here's a quick example of what the output to dependencies looks like.
So if you have some pair RDD words, and you do groupByKey and
then you call dependencies.
What you get back from the Spark is a list with shuffle dependencies in it.
So you know that this called to groupByKey is going to cause the shuffle.
Spark also comes with another method called toDebugString.
This method actually prints out a visualization of the RDD lineage along
with other information relevant to scheduling.
So if I was to use the same example on the previous slide, so we do map and
then groupByKey on some pair RDD of words, if we call toDebugString,
what we see is this.
The resulting RDD from group by key is a ShuffledRDD,
which came from a MapPartitionsRDD.
Which itself came from a ParallelCollectionRDD.
So this ParallelCollectionsRDD corresponds to wordsRdd.
This MapPartitionsRDD corresponds to the results of map on this wordsRdd.
And finally, ShuffledRDD corresponds to the result of groupByKey.
The indentations here actually show how Spark groups together these operations.
So operations in the same indentation, like for example MapPartitions here,
these groupings are typically separated by shuffles.
So you can print out a lineage using toDebugString, and
you'll be able to see how your job will be broken up into
different groupings of operations, separated by shuffles.
So for example you can have a group of many narrow transformations,
followed by one wide transformation.
And you can see that from the indentation from the lineage printed out,
from toDebugString.
And if you want to understand a little bit more about how Spark Jobs
are actually run, these groupings are called stages.
So a spark job is broken into stages by its scheduler.
So we saw these lineage graphs and
how they actually represent all of the relationships between RDDS,
and how we can determine when a shuffle will occur, but at the beginning
of this session, I made this claim that lineage is related to fault tolerance.
So how might that be the case?
Well, actually lineages are the key to fault tolerance in Spark.
Ideas for functional programming enable this fault tolerance in Spark.
This is because these are immutable, so
we can't update or change any of the data inside of an RDD.
And then by using higher-order functions like map, flatMap and
filter to do functional transformations on our immutable data,
what we have is effectively a Directed Acyclic Graph.
And assuming that we keep around this function that we passed to operations like
map, flatMap and filter, what we can do is we can re-compute at
any point in time any of the RDDs in our entire lineage graph.
So with these three aspects of Spark's design,
RDDs being immutable, everything being essentially higher order functions, and
then passing functions to these higher order functions, and building
a Directive Accyclic Graph, building a tree of transformation out of that.
This means that we can actually,
recover from failures by recomputing lost partitions from the lineage graphs.
So if you keep this information around.
If you save this somewhere to stable storage,
then you'll always be able to re-derive individual RDDs in your lineage graph.
That's the key idea here.
And that's how Spark achieves fault tolerance without having to write to disk.
This is how Spark can still do everything in memory, and also be fault tolerant.
Of course, this is always more easily understood visually.
So let's assume one of our partitions from the previous example fails.
So let's assume this partition here for some reason, is no longer any good.
Without having to have checkpoints and all that data to disk, Spark can actually just
re-drive it using this graph of dependencies that it all ready has.
For example, we know that this piece of data is derived from this piece of data,
which is derived from this piece of data.
Assuming that non of it is cached in memory, and I can't just reuse this, and
all we all have to do is recompute these pieces of data.
So all we have to do is go back to this dependency information along
with those functions that are stored with those dependencies, and just
recompute these pieces of data in which we compute that piece of data, like so.
And viola.
Now note, now we can put in this partitions is fast for
narrow dependencies, but it's slow for y dependencies.
But looking then at a visualization to see why that's the case.
If we lost this partition instead, we have to trace our dependencies backward.
And because this child dependency depends on all of the parent dependencies
in this union, we've gotta recompute all of these intermediate RDDs,
in order to recompute this one piece of data here.
Of course, it helps that this RDD is cached memory so
we can just reuse this partition in, without having to compute this groupBy.
But this is still going to be very expensive.
So losing partitions that were derived from a transformation with
wide dependencies, can be much slower.
