Today, we would like to talk about Spark
SQL Join Improvement at Facebook
more specifically, whether you're focused
on several join improvement we did
internally or around the spark platform at Facebook.
Before we start the talk,
just a brief introduction about myself.
I'm a software engineer working on the
data platform team at Facebook.
I'm a Spark contributor,
mostly focusing on Spark SQL. Previously,
I worked on the Hive and Hadoop team at Facebook.
Here is the agenda for this talk.
We will first start off with a brief overview of
Spark SQL Joins, and then talk about
our improvement around shuffled Hash Join.
After that, we will take a
brief discussion for ongoing work
to leverage bloom filter for Join.
Then, we will discuss the improvement
around the stream-stream Join, and
finally conclude by showing some of the problems
that we are working on in this space.
Let's take a brief overview of our Spark SQL Joins.
What are they, how they work, and when to use.
Spark SQL has three different drawing operators.
Broadcast hash join, shuffle hash join, and sort merge join.
Let's go over them one by one.
As an example, we want to join two data sources: A and B.
Broadcast hash join is to join a
super small table with another table.
Suppose data source A is pretty small,
Spark driver will build the hash relation
on top of source A and the broadcast A to every task.
Each task will read a portion of table B,
and hash join to look how to join with table A.
The requirement for broadcast hash join
is a data size of one table
should be smaller than the config.
Spark SQL auto broadcast joins threshold,
which is 10 megabytes by default.
The pros of broadcast hash join is
there is no shuffle and sort needed on both sides.
And it doesn't have any skew issues.
The cons is that if source A is not small enough,
it can cause auto memory, OOM on driver side, when
building a high stimulation.
Next is sort merge join. It first shuffles both
sides A and B on join key. And after a shuffle,
each task will do a local sort on both sides.
Now the join is performed in a sort merge way.
The pros of sort merge join is
it doesn't have any OOM issues and it
can handle large data sizes well on both sides.
So if you have two large tables to join,
sort merge join is a better choice.
The cons is that it needs to shuffle and
sorts before the join and this can be
inefficient and costly for a large table.
For example, the internal sort can be expensive
if the table is big and needs to skew auto memory.
In addition, if the join keys are skewed-,
for example, has a lot of nouns as join key,
sort merge join can have data skew problems.
The third one is shuffle hash join.
It first shuffles, both sides, A and B on join key.
And after shuffle, each task will build
hash table on smaller side and stream
the other side to form hash table hash join
similar to the broadcasted hash join.
One thing to know here is that join is disabled by
default, whereas spark SQL join prefers
SQL sort merge join config, and the ones
that should be smaller than spark SQL,
auto broadcast join's threshold, tend to
shuffle spark SQL's partitions.
Which is around the two gigabytes by default.
And the smallest size should be
3X smaller than the larger side.
The pros of shuffle hash join is that
it can handle large data as, as well, on one side.
It does not need sort compared to sort merge join.
And in some cases, shuffle join is more
efficient than sort merge join.
The cons are that it needs to shuffle before join.
And it has similar skill problem as sort merge join.
In addition, shuffle hash join has a unique challenge
that it can own on cast side for building hash table.
And we will talk more about how we
tackle this problem shortly.
Here, I hope everyone gets an idea of how spark SQL joins.
Here, we talk about the improvement we
did around the shuffle hash join.
The first one, is to end code-gen
support for shuffle hash join.
The motivation is to improve CPU computer resourcing
by leveraging spark SQL hosting the code-gen.
Previously, only broadcast hash join and
soft merge join had code-gen support.
Here, we add the code-gen support
for shuffle hash join as well.
How do we do that? For implementation,
we refactor broadcast hash join and code-gen logic
into common parent class for HashJoin.scala.
As broadcast hash join and shuffle hash join
has similar logic to logic to high table look up join.
For benchmark inquiry, we are seeing
30% of run-time improvement compared
to non-code-gen code pass.
And the PR is merged into our stream and
it will be available in spark 3.1.
The second one is to end the Full Outer join support.
The motivation is to improve CPO and IO.
Previously, only sort merge join
supports full outer join,
and the sort can be very expensive
when table is large and needs to spill to disk.
Shuffle hash join as the other hand does not need sort.
And here is how we do that, for the join
not only needs to output all the matched rows
from both sides, but also needs
to output all non-matched rows.
And for stream side, this is a trivial problem.
As for each row, will build a hash table look up.
We know exactly whether the row is measured immediately,
but it's non-trivial for build side.
For each row in hash table build side,
we only know whether the row has a match or not
after exhausting every row from the other stream side.
So we need actual data structure
to recall this information for build side.
And here, we use hash side to recall
if the rows have been met or not.
Here is a figure for how full
outer shuffle hash join works.
First, we do shuffle on join keys on both sides.
That would build hash table on smaller side for each task.
When joining each row from stream side,
we maintained a separate hash set data structure
to recall the matches rows from build side.
After reading all rows from stream side,
we go over all rows from hash table again,
and all rows that are not matched rows.
For Benchmark query,
We are seeing 30% of runtime improvement
compared to Full outer shuffle join.
And the PR is merged into screen our stream
and will be available in sparks 3.1.
The third one is to end sort-based fallback
mechanism for shuffle hash join.
The major fail point for shuffle hash join
is the OOM on task side when
building the hash table.
There's no fallback and no spill logic for the
hash table right now for the shuffle hash join.
Once there is not enough memory for building the table,
the task will be filled immediately and the query
will be filled immediately in hardware.
Okay. With this OOM limitation,
shuffle hash join is hard to enable by default.
And how we can tackle this problem?
Here we introduce fallback logic,
when building hash table.
Whenever we fail to get memory to insert
current role to house table, we stop here,
and sort both streams on build side.
And do sort merge join, instead of shuffle hash join.
The fallback is dynamic during the downtime result and in
conflict tuning. And then,
we enable this feature by default
at Facebook for several years.
We don't see OOM problem for shuffle hash join after
ending this fallback feature. And,
the PR is Work in Progress and
hopefully it can be available in sparks 3.1.
Okay, here we go to the next topic.
To leverage pro filter for joint operators.
Bloom filter is a data structure to test the membership.
For a given element, it answers the question,
whether it's in a set of elements or not.
The motivation is to use the bloom filter,
to filter all the rows as early as possible to save CPU.
And I already saw it for shuffle hash
join and sort merge join.
First, the bloom filter will be appealed on
join keys of smaller size.
Now the bloom filter will be broadcasted to every
task when scanning the larger side.
In this way, it reduces the amount of data
to process in followed shuffle and sort stage.
The design is under discussion with several community
members and we will submit the JIRA with design later.
The last topic we want to discuss is
a stream-stream join improvement
we worked on in spark structured streaming.
Here is a quick refreshment for stream-stream join.
There is one join operator, for
stream-stream join, which is called
streaming symmetric hash join exec.
And here's how it works as an illustration.
For structured streaming macro batch execution
mode in each micro batch, the
source A and B are shuffled on join keys
and each macro batch will join
with a state store on the other side,
where a hash table look up becomes join mechanism.
The state store basically contains rows
from parent patches on the other side.
After the join is done, the current macro
batch of data will be stored into each system.
And this is for this macro batch to join.
Normally, the watermark in stream-stream join
is used in SQL query to remove the
stale rows from state store to where state
store starts growing infinitely.
And our first work is to, have the
left semi join support into stream-stream join.
And the motivation is, we found actually
for some FB streaming workloads, left's semi join
is even more popular than left outer join.
And here's an example.
We have a, suppose we have a stream of data
of ads impression and ads the impression,
you can think of it ads joined to the users
and the another stream of data of ads click,
which means the ad has a click from the user.
And we want to get all the ads impression
which has ads clicked from the user,
by the way, its not clear what those clicks are.
So, we only need to match it ads the impression
and the match clicks.
And as this is exactly in that semi join semantics,
and if we do it with left outer join,
it will be very verbose as we
output all the ads impression looks.
And how we do the left semi join? Okay.
So let's dive into the role for each row.
We check if there's a match on the right side state store.
And if there is a match,
we output the left side row immediately
because there is a match for the last semi join
semantics and the pathway does not need to put the left
side row, the matched row, into left side
state store because there's no need to put the
state store, anyway. If was there no match,
and we do not output anything then,
put the row into left state store.
And when putting this into left state store,
we mark the matched field to set to force into state store.
Its the same way for the right side state row.
We check if there's a match on left side state store.
For all matched left rows in state store,
we output the rows with matched field as false.
So this means we only output the left side rows,
that matched for the first time to
guarantee left semi joined semantics.
And after that, we set all the left rows
with matched field to be true,
and it will not be joined and output again.
The state store eviction logic is to
evict rows from false left and right side
state store below the watermark.
There is no customer logic around that.
And in that way, its the same as inner join watermark logic.
And then for the PR,
it's merged into our stream already
and it will be available in spark 3.1.
The second one is to add the full outer join support.
Stream-stream join already supports
left outer and right outer join.
And the full outer join, it's kind of trivial in this case,
because it is basically a combination between
left outer and right outer join.
Now, how can we achieve that?
For left side input row, we check
if there's a match for right side state store.
And if there's a match, we output
all the matched rows immediately
and put the row in the left side state store.
For the right side input row, we check
if there's a match in left side state store.
And if there's a match output all matched rows
and update left side row state
with matched field to set to true.
And then we also put all the rows in this match
into right side state store.
For left sides, rows need to be evicted
from state store but only output rows
if it's a non-matched row,
because all the matched rows
has been output already.
And this is the same
state store eviction logic for right side.
The PR is work in progress and hopefully,
it can be available in sparks 3.1.
And to recap, in this session with
the three major points.
First, we go over spark SQL join implementations.
Then, we talked about the shuffle hash join
improvement we did internally.
And then we briefly discussed,
to leverage the bloom filter
for the shuffle join and sort merge join.
In the end, just saw some of the improvement
we are working on for the stream-stream join
instruction streaming improvements.
For future work in this area, we are working
on history based optimization to select
best joint strategy at the wrong time.
What observation in today's data warehouse is normally a
data engineer and a software engineer
that data pipelines around them everyday daily.
So mostly of the SQL queries, I repeated it every day.
Just with different equal improvisations every day.
So with this optimization, we can say like
pre-teach today's query runtime metrics
with historical query runtime metrics. Such as
join data size and with join data size,
we can decide which joint to use.
And that's all. And this concludes my talk.
Feel free to ask any questions and your feedback
is very important to us. Thank you.