- Hello, everyone.
I'm very glad here, to speak at this summit.
And, my topic is improving Spark SQL performance by 13%.
How we optimize?
How could it fail to approach start?
And the Parquet Reader.
I'm Sun Ke, and I'm senior engineer
of Data Engine team from ByteDance.
First, let me introduce who we are, and what we do.
We are Data Engine team of ByteDance,
and we build a platform of one-stop experience for OLAP,
on which users can analyze EB level data
by writing SQL without caring about
the underlying execution engine,
or the (mumbles) Spark SQL, Presto, or Hive.
We manage Spark SQL, Presto, Hive workload,
and we offer open API and serverless OLAP platform
for users in ByteDance.
At the same time, we optimize Spark SQL, Presto, Hudi,
and Hive engine in deep.
Also, we design data architecture
for most business lines in ByteDance.
Okay, next up is my agenda.
There are three parts.
First, Spark SQL and ByteDance.
Second, how Spark SQL read Parquet.
In this part, I will introduce what we can do
to improve Spark SQL and the Parquet performance.
Third, optimization detail
of Parquet Filter Pushdown and Parquet Reader at ByteDance.
The first part: Spark SQL at ByteDance.
In 2016 we started to use Spark SQL
in small scale experiments.
In the next year, we use the Spark SQL in ad-hoc workload.
And in 2018, we started to use Spark SQL
in few ETL workload.
In 2019, we deployed the Spark SQL
in full production and started to migrate
from Hive to Spark SQL.
In this year, Spark SQL is
the main engine in data warehouse.
By now, Spark SQL covers 98% ETL workload in ByteDance,
and Parquet is default file format in data warehouse,
and vectorizedReader is also enabled by default.
(mumbles).
vectorizedReader can improve Spark's performance greatly.
In many cases, the rate is 50%.
The next part: How Spark Read Parquet.
This part is divided into three sub-parts.
First, overview of Parquet.
Second, procedure of Parquet reading.
Third, what we can do to optimize.
First, overview of Parquet, and show in the right figure.
This is the layout of Parquet from the official website.
And, we can see Parquet file is divided
into many work groups and a single folder.
And is a work group and a folder.
How is they are?
Statistics, for example.
And, Spark and Parquet can skip useless data
by filter pushdown using Footer and RowGroup statistics.
For example, if the maximum of the work group is 10,
and the rate wanted to acquire data greater
than 20, Parquet can skip this work group by the statistics.
Okay, and we all know Parquet
is a commonly known storage format,
so it's easy to do column pruning
and it's more efficient to compress data in Parquet file.
Next, I will introduce the procedure
of Parquet reading in Spark SQL.
The entry is DataSourceScanExecution-class,
and the list class in work; ParquetFileFormat,
and the ParquetFileFormat in work.
VectorizedParquetReactorReader,
and this class is the most important class
in Parquet reading.
VectorizedParquetReactorReader skips useless RowGroups
by filter pushdown translated by ParquetFilters,
and VectorizedParquetReactorReader builds column readers
for every target column,
and these column readers read data together in batch,
like one-by-one.
For example, we wanted to read the three columns.
The VectorizedParquetReactorReader, we reviewed
the three column reader.
Okay, next I will give a circle example
to show how Spark reads Parquet.
The SQL is, "select star from table name
where date equals some date, and the category equals test."
In the table, data is partition column,
and the category is predicate column,
and the Parquet file is the Parquet file.
And there is a statistic of the Parquet file
is in the voting chart.
For example, the min. num. of category in RowGroup1
is a1, and the max. num. of category in RowGroup1
is z1, so, when needed to, read it as a RowGroup
because test is in the (mumbles) of this work group.
By the same logic, we need to read work group 2
and work group 3.
So, Spark reads all the three work group
because the statistics of the three work groups
are not distinguishable.
In proper production environment,
Parquet filter pushdown works poorly
because predicate columns are out of order
in Parquet files, and this phenomenon is reasonable.
So, it's very valuable to sort
the common used predicate columns
in Parquet files to reduce IO.
Next, I will give another second example.
The circle is selected to column 1 from table name,
where date equals some date,
and columns two equals test.
And, Parquet file is voting figure.
RowGroup1 is skipped by filter pushdown,
and column 3 is skipped by column pruning,
because column 3 is not involved in the SQL.
So, columns one and column two are read together
by vectorizedReader.
In many cases, if the first read
of the extraction column two equals test is very high,
then most the data of column one is unnecessary to read.
So, it is very valuable to read and filter data
by filter columns first, and it then skips useless data
in the according row number,
and only reads the useful data of other columns.
So, I have finished the optimization introduction,
and now the optimization detail
of Parquet Filter pushdown and Parquet Reader at ByteDance.
First, we finding that statistics of Parquet
are not distinguishable,
so we wanted to increase Parquet statistics discrimination.
But, our users have some requirements.
They want low overhead.
Sort all data is expensive or even impossible,
so we can only sort every single file.
And, also, they want automation,
because they don't want to update every ETL jobs.
So, we're propose a feature named LocalSort
which add a sort execution node
before InsertIntoHiveTable node.
In this feature, Spark will sort some columns
in every single Parquet file to increase Parquet
and statistics discrimination.
(mumbles), the execution plan of Spark SQL is as below,
and we can see a sort execution node
is added before inserting into HiveTable node.
How come the question, which columns to be sorted?
And our strategy is to analyze the history queries
and choose the most common used predicate columns,
and then configure sort columns to table property
of hive table and Spark SQL will read
this property during execution time.
The procedure is automatic without manual intervention,
which meets the user's requirements.
Just like the previous example,
after sorting data by column category,
the statistics of Parquet is in the voting chart.
For example, in the min. num.
of category column in RowGroup1 is a1,
and the max. num. of category in RowGroup1 is g1,
so we don't need to read a literal group,
and of the min. num. of category column in RowGroup2,
it's g2, and the max. num. of category in RowGroup2 is u2.
So, we need to read this row group.
And, the min. num. of category in RowGroup3 is u3,
and the max. num. of category column in RowGroup3 is z3.
We don't need to read this RowGroup2, either.
So, after using (mumbles), Spark is only one RowGroup,
and this feature would use IO greatly.
By using (mumbles), Spark reads less data
for more efficient statistics,
and Parquet file size is much smaller
for data is more efficient to compress.
Because we only sort columns in every single file,
the overhead is only 5% in production experiments.
Okay, I have finished the optimization detail
of Parquet filter pushdown.
Next, I will introduce optimization detail
of Parquet Reader.
And, we find that Spark reads too much unnecessary data,
so we wanted to...
we needed to filter unnecessary data as soon as possible.
So, we proposed a feature named the Prewhere.
And this idea is from ClickHouse,
which is also famous OLAP platform.
In Prewhere, Spark read data of filter columns
in batch firstly, and then will skip other columns
if the data is unmatched
with filter columns in the same row.
And then, we only read the data which is matched
with filter column in the same row.
And, how we achieve this feature?
We split Parquet reader into two readers.
First, the filter reader for filter column,
and the non-filter reader for other columns.
As we can see, in the figure.
Next is a procedure of Parquet reading.
First, we use filter reader to read data in batch,
and then apply filter expressions to data.
If unmatched, we continue to use filter reader
to read data until it's matched.
If it's matched, we use non-filter reader
to read data in batch and skip useless data in the same row.
In the end, we will union the data from filter reader and,
after this procedure reader will show great benefit.
In some cases, read can skip Parquet page.
In some case, read can skip Parquet coding.
In some case, we can even skip Parquet RowGroup.
In our production environment, the commonly used types
of body.
For example, ByteType, ShortType, LongType, DoubleType,
and the most commonly used filter type also is body.
For example, greater, greater equal,
in, is null, is not null.
After using (mumbles) and the premier, rate achieved
a great benefit, and the rate improve Parquet performance
by 13% in many cases.
Okay, that's all.
This end of my session.
Thank you.