(DEPRECATED) Apache Flink Mailing List archive.

[DISCUSS] Vectorization Support in Flink

Classic

List

Threaded

8 messages Options

Fan Liya

[DISCUSS] Vectorization Support in Flink

Hi all,

We have opened an issue about vectorization in Flink (FLINK-13053
<https://issues.apache.org/jira/browse/FLINK-13053>). Would you please give
your valuable feedback? Thank you in advance.

Vectorization is a popular technique in SQL engines today. Compared with
traditional row-based approach, it has some distinct advantages, for
example:

1) Better use of CPU resources (e.g. SIMD)

2) More compact memory layout

3) More friendly to compressed data format.

Currently, Flink is based on a row-based SQL engine for both stream and
batch workloads. To enjoy the above benefits, we want to bring
vectorization to Flink. This involves substantial changes to the existing
code base. Therefore, we give a plan to carry out such changes in small,
incremental steps, in order not to affect existing features. We want to
apply it to batch workload first. The details can be found in our proposal.

For the past months, we have developed an initial implementation of the
above ideas. Initial performance evaluations on TPC-H benchmarks show that
substantial performance improvements can be obtained by vectorization (see
the figure below). More details can be found in our proposal.

[image:
https://lh5.googleusercontent.com/hjXkXGImWOjaiB8zF0SKIMoItY6VCBm-BmJWWEXRo0ZPHdwLgKzCmIoNKef1YPCaAA7NXN6RvO-nwBBXBee52KeAtBjyIvh_NcAuChvW3BEtQuZGL5GPddqxL_iMV7HvEVCC6k-m]

Special thanks to @Kurt Young’s team for all the kind help.

Special thanks to @Piotr Nowojski for all the valuable feedback and help
suggestions.

Best,

Liya Fan

Fan Liya

Re: [DISCUSS] Vectorization Support in Flink

Performance chart. FYI.

Best,

Liya Fan

On Tue, Jul 2, 2019 at 3:37 PM Fan Liya <[hidden email]> wrote:

Hi all,

We have opened an issue about vectorization in Flink (FLINK-13053). Would you please give your valuable feedback? Thank you in advance.

Vectorization is a popular technique in SQL engines today. Compared with traditional row-based approach, it has some distinct advantages, for example:

1)      Better use of CPU resources (e.g. SIMD)

2)      More compact memory layout

3)      More friendly to compressed data format.

Currently, Flink is based on a row-based SQL engine for both stream and batch workloads. To enjoy the above benefits, we want to bring vectorization to Flink. This involves substantial changes to the existing code base. Therefore, we give a plan to carry out such changes in small, incremental steps, in order not to affect existing features. We want to apply it to batch workload first. The details can be found in our proposal.

For the past months, we have developed an initial implementation of the above ideas. Initial performance evaluations on TPC-H benchmarks show that substantial performance improvements can be obtained by vectorization (see the figure below). More details can be found in our proposal.

Special thanks to @Kurt Young’s team for all the kind help.

Special thanks to @Piotr Nowojski for all the valuable feedback and help suggestions.

Best,
Liya Fan

Jeff Zhang

Re: [DISCUSS] Vectorization Support in Flink

Hi Liya,

Displaying image is not supported in apache mail list, you need to put it
elsewhere and post link in mail list.

Fan Liya <[hidden email]> 于2019年7月2日周二下午3:40写道：

> Performance chart. FYI.
>
> Best,
> Liya Fan
> [image: image.png]
>
> On Tue, Jul 2, 2019 at 3:37 PM Fan Liya <[hidden email]> wrote:
>
>> Hi all,
>>
>> We have opened an issue about vectorization in Flink (FLINK-13053
>> <https://issues.apache.org/jira/browse/FLINK-13053>). Would you please
>> give your valuable feedback? Thank you in advance.
>>
>> Vectorization is a popular technique in SQL engines today. Compared with
>> traditional row-based approach, it has some distinct advantages, for
>> example:
>>
>>
>>
>> 1) Better use of CPU resources (e.g. SIMD)
>>
>> 2) More compact memory layout
>>
>> 3) More friendly to compressed data format.
>>
>>
>>
>> Currently, Flink is based on a row-based SQL engine for both stream and
>> batch workloads. To enjoy the above benefits, we want to bring
>> vectorization to Flink. This involves substantial changes to the existing
>> code base. Therefore, we give a plan to carry out such changes in small,
>> incremental steps, in order not to affect existing features. We want to
>> apply it to batch workload first. The details can be found in our proposal.
>>
>>
>>
>> For the past months, we have developed an initial implementation of the
>> above ideas. Initial performance evaluations on TPC-H benchmarks show that
>> substantial performance improvements can be obtained by vectorization (see
>> the figure below). More details can be found in our proposal.
>>
>>
>>
>> [image:
>> https://lh5.googleusercontent.com/hjXkXGImWOjaiB8zF0SKIMoItY6VCBm-BmJWWEXRo0ZPHdwLgKzCmIoNKef1YPCaAA7NXN6RvO-nwBBXBee52KeAtBjyIvh_NcAuChvW3BEtQuZGL5GPddqxL_iMV7HvEVCC6k-m]
>>
>>
>>
>> Special thanks to @Kurt Young’s team for all the kind help.
>>
>> Special thanks to @Piotr Nowojski for all the valuable feedback and help
>> suggestions.
>>
>>
>> Best,
>>
>> Liya Fan
>>
>

--
Best Regards

Jeff Zhang

Ji Liu

Re: [DISCUSS] Vectorization Support in Flink

Hi Liya,
Thanks for opening this discuss.
+1 for this, vectorization makes sense for Flink especially for batch work loads, I think Flink should look into supporting it progressively.

Thanks,
Ji Liu

------------------------------------------------------------------
From:Jeff Zhang <[hidden email]>
Send Time:2019年7月2日(星期二) 15:50
To:dev <[hidden email]>
Subject:Re: [DISCUSS] Vectorization Support in Flink

Hi Liya,

Displaying image is not supported in apache mail list, you need to put it
elsewhere and post link in mail list.

Fan Liya <[hidden email]> 于2019年7月2日周二下午3:40写道：

--
Best Regards

Jeff Zhang

Fan Liya

Re: [DISCUSS] Vectorization Support in Flink

In reply to this post by Jeff Zhang

@Jeff Zhang, thanks for your kind reminder.
The image can be found from our proposal
<https://docs.google.com/document/d/1cUHb-_Pbe4NMU3Igwt4tytEmI66jQxev00IL99e2wFY/edit#heading=h.50xdeg1htedb>
or the JIRA <https://issues.apache.org/jira/browse/FLINK-13053>.

Best,
Liya Fan

On Tue, Jul 2, 2019 at 3:49 PM Jeff Zhang <[hidden email]> wrote:

> Hi Liya,
>
> Displaying image is not supported in apache mail list, you need to put it
> elsewhere and post link in mail list.
>
>
>
> Fan Liya <[hidden email]> 于2019年7月2日周二下午3:40写道：
>
> > Performance chart. FYI.
> >
> > Best,
> > Liya Fan
> > [image: image.png]
> >
> > On Tue, Jul 2, 2019 at 3:37 PM Fan Liya <[hidden email]> wrote:
> >
> >> Hi all,
> >>
> >> We have opened an issue about vectorization in Flink (FLINK-13053
> >> <https://issues.apache.org/jira/browse/FLINK-13053>). Would you please
> >> give your valuable feedback? Thank you in advance.
> >>
> >> Vectorization is a popular technique in SQL engines today. Compared with
> >> traditional row-based approach, it has some distinct advantages, for
> >> example:
> >>
> >>
> >>
> >> 1) Better use of CPU resources (e.g. SIMD)
> >>
> >> 2) More compact memory layout
> >>
> >> 3) More friendly to compressed data format.
> >>
> >>
> >>
> >> Currently, Flink is based on a row-based SQL engine for both stream and
> >> batch workloads. To enjoy the above benefits, we want to bring
> >> vectorization to Flink. This involves substantial changes to the
> existing
> >> code base. Therefore, we give a plan to carry out such changes in small,
> >> incremental steps, in order not to affect existing features. We want to
> >> apply it to batch workload first. The details can be found in our
> proposal.
> >>
> >>
> >>
> >> For the past months, we have developed an initial implementation of the
> >> above ideas. Initial performance evaluations on TPC-H benchmarks show
> that
> >> substantial performance improvements can be obtained by vectorization
> (see
> >> the figure below). More details can be found in our proposal.
> >>
> >>
> >>
> >> [image:
> >>
> https://lh5.googleusercontent.com/hjXkXGImWOjaiB8zF0SKIMoItY6VCBm-BmJWWEXRo0ZPHdwLgKzCmIoNKef1YPCaAA7NXN6RvO-nwBBXBee52KeAtBjyIvh_NcAuChvW3BEtQuZGL5GPddqxL_iMV7HvEVCC6k-m
> ]
> >>
> >>
> >>
> >> Special thanks to @Kurt Young’s team for all the kind help.
> >>
> >> Special thanks to @Piotr Nowojski for all the valuable feedback and help
> >> suggestions.
> >>
> >>
> >> Best,
> >>
> >> Liya Fan
> >>
> >
>
> --
> Best Regards
>
> Jeff Zhang
>

Fan Liya

Re: [DISCUSS] Vectorization Support in Flink

In reply to this post by Ji Liu

@Ji Liu, thanks a lot for your feedback.
This work must be performed in a progressive manner, so as not to break
existing code.

Best,
Liya Fan

On Tue, Jul 2, 2019 at 3:57 PM Ji Liu <[hidden email]> wrote:

> Hi Liya,
> Thanks for opening this discuss.
> +1 for this, vectorization makes sense for Flink especially for batch work
> loads, I think Flink should look into supporting it progressively.
>
> Thanks,
> Ji Liu
>
>
> ------------------------------------------------------------------
> From:Jeff Zhang <[hidden email]>
> Send Time:2019年7月2日(星期二) 15:50
> To:dev <[hidden email]>
> Subject:Re: [DISCUSS] Vectorization Support in Flink
>
> Hi Liya,
>
> Displaying image is not supported in apache mail list, you need to put it
> elsewhere and post link in mail list.
>
>
>
> Fan Liya <[hidden email]> 于2019年7月2日周二下午3:40写道：
>
> > Performance chart. FYI.
> >
> > Best,
> > Liya Fan
> > [image: image.png]
> >
> > On Tue, Jul 2, 2019 at 3:37 PM Fan Liya <[hidden email]> wrote:
> >
> >> Hi all,
> >>
> >> We have opened an issue about vectorization in Flink (FLINK-13053
> >> <https://issues.apache.org/jira/browse/FLINK-13053>). Would you please
> >> give your valuable feedback? Thank you in advance.
> >>
> >> Vectorization is a popular technique in SQL engines today. Compared with
> >> traditional row-based approach, it has some distinct advantages, for
> >> example:
> >>
> >>
> >>
> >> 1) Better use of CPU resources (e.g. SIMD)
> >>
> >> 2) More compact memory layout
> >>
> >> 3) More friendly to compressed data format.
> >>
> >>
> >>
> >> Currently, Flink is based on a row-based SQL engine for both stream and
> >> batch workloads. To enjoy the above benefits, we want to bring
> >> vectorization to Flink. This involves substantial changes to the
> existing
> >> code base. Therefore, we give a plan to carry out such changes in small,
> >> incremental steps, in order not to affect existing features. We want to
> >> apply it to batch workload first. The details can be found in our
> proposal.
> >>
> >>
> >>
> >> For the past months, we have developed an initial implementation of the
> >> above ideas. Initial performance evaluations on TPC-H benchmarks show
> that
> >> substantial performance improvements can be obtained by vectorization
> (see
> >> the figure below). More details can be found in our proposal.
> >>
> >>
> >>
> >> [image:
> >>
> https://lh5.googleusercontent.com/hjXkXGImWOjaiB8zF0SKIMoItY6VCBm-BmJWWEXRo0ZPHdwLgKzCmIoNKef1YPCaAA7NXN6RvO-nwBBXBee52KeAtBjyIvh_NcAuChvW3BEtQuZGL5GPddqxL_iMV7HvEVCC6k-m
> ]
> >>
> >>
> >>
> >> Special thanks to @Kurt Young’s team for all the kind help.
> >>
> >> Special thanks to @Piotr Nowojski for all the valuable feedback and help
> >> suggestions.
> >>
> >>
> >> Best,
> >>
> >> Liya Fan
> >>
> >
>
> --
> Best Regards
>
> Jeff Zhang
>

JingsongLee-2

Re: [DISCUSS] Vectorization Support in Flink

Thank liya for bringing this discussion.
As the changes are very big and has a great impact on architecture.
I think we should be more clear about the benchmark. We need to
be cautious about testing to ensure that we really benefit from it.

About the benchmark:
1.Can you make the test report clearer? For example, environment
and data scale.
2.Can test scenarios be richer? For example, benchmark about
spill scenarios. Benchmark about TPCDS.
3.Can we have more detailed test conclusions? About what kind of
case will be quicker. At present, the calculation of blink-planner is
not perfect. For example, it can avoid the overhead of virtual
function calls. Aggregate algorithm needs to be improved. Can
you can make further analysis with your benchmark.

> More compact memory layout
I think BinaryRow and ColumnarRow already have an efficient
and compact memory layout.

Just like you mentioned in doc. Blink's ColumnarRow now has
vector computing features. And we can also push down a lot of
calculations into the specific source, which can be more native
to support the calculation near the source. I don't think complete
vector calculation is that necessary. Because of the following
reasons, the latter calculation is difficult to obtain benefits through
Vector calculation:
1. Maybe the cost of conversion between VectorBatch and Row will
be the performance killer. I think maybe we should do some
performance test to it. If there are join/shuffle nodes, there will be
vector-to-row and row-to-vector overhead? These two operators
are often the key to job performance.
2. Operators like sort, aggregation, their vectorized computational
versions maybe need more benchmarks. I have no idea about it.
3. Now Java SIMD can only improve a limited number of vector
computation like filter and calc, but often the bottleneck of batch
jobs is not there, more on Join and Shuffler. Complete Java vector
computing looks like a long way off. If we vectorize through JNI,
the cost of JNI can not be ignored. And SIMD algorithm is not
necessarily faster, it brings a lot of additional data copies.
4. If we move forward with CodGenerator(Like Spark
WholeStageCodeGen), can we achieve better results without
vector computation? The JavaCompiler/JVM may optimize the
code to SIMD.

Other thing is that the vector version of operators maybe need
consider the problem of memory management?

Best, JingsongLee

------------------------------------------------------------------
From:Fan Liya <[hidden email]>
Send Time:2019年7月2日(星期二) 16:31
To:dev <[hidden email]>; Ji Liu <[hidden email]>
Subject:Re: [DISCUSS] Vectorization Support in Flink

@Ji Liu, thanks a lot for your feedback.
This work must be performed in a progressive manner, so as not to break
existing code.

Best,
Liya Fan

On Tue, Jul 2, 2019 at 3:57 PM Ji Liu <[hidden email]> wrote:

Fan Liya

Re: [DISCUSS] Vectorization Support in Flink

Hi JingsongLee,

Thanks a lot for your feedback. It's nice discussing with you guys :-)

My comments below:

1. Sorry I did not clearly state the experimental environment. The
evaluations were performed on a cluster with 21 N42 nodes (64 core, 256 GB
memory, 10 Gb/s network card). I will add such information in our doc. The
dataset is clearly stated in our proposal: 1TB TPC-H.

2. We emphasize that this is only initial implementation, and the
performance results are initial results. That means our work is far from
perfect. So invite the Flink community to work with us, to make it perfect.

3. Can you please give more details about blink-planner avoiding virtual
function calls? I would like to know more about that.

4. I agree that BinaryRow and ColumnarRow are highly efficient, compared
with previous row data structures in Flink. However, my point is that, they
can be further improved. Let's talk about BinaryRow first: Each field with
fixed width takes 8 bytes, right? That means 50% overhead for an int, 75%
overhead for a short/char, 87.5% overhead for a byte, and 98% overhead for
a boolean. The memory layout of BinaryRow is far from being perfectly
compact.

5. Then let's talk about ColumnarRow. It is closer to the perfect compact
memory layout. But, it is not perfect either. The limitations of
ColumnarRow include: 1) It is based on heap space, the problem with this
can be found in [1]. 2) Its memory layout can be further improved, e.g. by
using a bit for a nullablility. 3) Lack of operations (e.g. realloc).

6. Calculation push down sounds like a great idea. What is the relation to
vectorization?

7. We agree that row/batch conversion is a performance killer. With the
current row-based implementation, such conversion is unavoidable, for input
formats like orc and parquet, right? Vectorization eliminates such
conversions completely.

8. SIMD can be applied to a wide range of operations, not just filter &
calc. For example, in hash join & hash agg, SIMD can be used to accelerate
the calculation of hash codes. In addition, if we use a row-based memory
layout, additional data copy is required, and that is why we use
vectorization, right?

9. Whole stage code-gen seems like an orthogonal technique to
vectorization. It is feasible to achieve the benefits of both whole-stage
code-gen and vectorization.

10. Memory management is an important problem. Fortunately, Flink provides
interfaces for off-heap memory management. So Arrow memory buffer can be
integrated with little effort.

As indicated in our proposal, if the users do not like vectorization, they
can simply turn it off with a single flag. So at the very least,
vectorization will not bring harm to Flink. It just enrich the SQL engine
for Flink, making it perform well for an even wider range of scenarios.
Just think about it.

Best,
Liya Fan

[1] Off-heap Memory in Apache Flink and the curious JIT compiler.
https://flink.apache.org/news/2015/09/16/off-heap-memory.html

On Tue, Jul 2, 2019 at 8:29 PM JingsongLee <[hidden email]>
wrote:

> Thank liya for bringing this discussion.
> As the changes are very big and has a great impact on architecture.
> I think we should be more clear about the benchmark. We need to
> be cautious about testing to ensure that we really benefit from it.
>
> About the benchmark:
> 1.Can you make the test report clearer? For example, environment
> and data scale.
> 2.Can test scenarios be richer? For example, benchmark about
> spill scenarios. Benchmark about TPCDS.
> 3.Can we have more detailed test conclusions? About what kind of
> case will be quicker. At present, the calculation of blink-planner is
> not perfect. For example, it can avoid the overhead of virtual
> function calls. Aggregate algorithm needs to be improved. Can
> you can make further analysis with your benchmark.
>
> > More compact memory layout
> I think BinaryRow and ColumnarRow already have an efficient
> and compact memory layout.
>
> Just like you mentioned in doc. Blink's ColumnarRow now has
> vector computing features. And we can also push down a lot of
> calculations into the specific source, which can be more native
> to support the calculation near the source. I don't think complete
> vector calculation is that necessary. Because of the following
> reasons, the latter calculation is difficult to obtain benefits through
> Vector calculation:
> 1. Maybe the cost of conversion between VectorBatch and Row will
> be the performance killer. I think maybe we should do some
> performance test to it. If there are join/shuffle nodes, there will be
> vector-to-row and row-to-vector overhead? These two operators
> are often the key to job performance.
> 2. Operators like sort, aggregation, their vectorized computational
> versions maybe need more benchmarks. I have no idea about it.
> 3. Now Java SIMD can only improve a limited number of vector
> computation like filter and calc, but often the bottleneck of batch
> jobs is not there, more on Join and Shuffler. Complete Java vector
> computing looks like a long way off. If we vectorize through JNI,
> the cost of JNI can not be ignored. And SIMD algorithm is not
> necessarily faster, it brings a lot of additional data copies.
> 4. If we move forward with CodGenerator(Like Spark
> WholeStageCodeGen), can we achieve better results without
> vector computation? The JavaCompiler/JVM may optimize the
> code to SIMD.
>
> Other thing is that the vector version of operators maybe need
> consider the problem of memory management?
>
> Best, JingsongLee
>
>
> ------------------------------------------------------------------
> From:Fan Liya <[hidden email]>
> Send Time:2019年7月2日(星期二) 16:31
> To:dev <[hidden email]>; Ji Liu <[hidden email]>
> Subject:Re: [DISCUSS] Vectorization Support in Flink
>
> @Ji Liu, thanks a lot for your feedback.
> This work must be performed in a progressive manner, so as not to break
> existing code.
>
> Best,
> Liya Fan
>
> On Tue, Jul 2, 2019 at 3:57 PM Ji Liu <[hidden email]> wrote:
>
> > Hi Liya,
> > Thanks for opening this discuss.
> > +1 for this, vectorization makes sense for Flink especially for batch
> work
> > loads, I think Flink should look into supporting it progressively.
> >
> > Thanks,
> > Ji Liu
> >
> >
> > ------------------------------------------------------------------
> > From:Jeff Zhang <[hidden email]>
> > Send Time:2019年7月2日(星期二) 15:50
> > To:dev <[hidden email]>
> > Subject:Re: [DISCUSS] Vectorization Support in Flink
> >
> > Hi Liya,
> >
> > Displaying image is not supported in apache mail list, you need to put it
> > elsewhere and post link in mail list.
> >
> >
> >
> > Fan Liya <[hidden email]> 于2019年7月2日周二下午3:40写道：
> >
> > > Performance chart. FYI.
> > >
> > > Best,
> > > Liya Fan
> > > [image: image.png]
> > >
> > > On Tue, Jul 2, 2019 at 3:37 PM Fan Liya <[hidden email]> wrote:
> > >
> > >> Hi all,
> > >>
> > >> We have opened an issue about vectorization in Flink (FLINK-13053
> > >> <https://issues.apache.org/jira/browse/FLINK-13053>). Would you
> please
> > >> give your valuable feedback? Thank you in advance.
> > >>
> > >> Vectorization is a popular technique in SQL engines today. Compared
> with
> > >> traditional row-based approach, it has some distinct advantages, for
> > >> example:
> > >>
> > >>
> > >>
> > >> 1) Better use of CPU resources (e.g. SIMD)
> > >>
> > >> 2) More compact memory layout
> > >>
> > >> 3) More friendly to compressed data format.
> > >>
> > >>
> > >>
> > >> Currently, Flink is based on a row-based SQL engine for both stream
> and
> > >> batch workloads. To enjoy the above benefits, we want to bring
> > >> vectorization to Flink. This involves substantial changes to the
> > existing
> > >> code base. Therefore, we give a plan to carry out such changes in
> small,
> > >> incremental steps, in order not to affect existing features. We want
> to
> > >> apply it to batch workload first. The details can be found in our
> > proposal.
> > >>
> > >>
> > >>
> > >> For the past months, we have developed an initial implementation of
> the
> > >> above ideas. Initial performance evaluations on TPC-H benchmarks show
> > that
> > >> substantial performance improvements can be obtained by vectorization
> > (see
> > >> the figure below). More details can be found in our proposal.
> > >>
> > >>
> > >>
> > >> [image:
> > >>
> >
> https://lh5.googleusercontent.com/hjXkXGImWOjaiB8zF0SKIMoItY6VCBm-BmJWWEXRo0ZPHdwLgKzCmIoNKef1YPCaAA7NXN6RvO-nwBBXBee52KeAtBjyIvh_NcAuChvW3BEtQuZGL5GPddqxL_iMV7HvEVCC6k-m
> > ]
> > >>
> > >>
> > >>
> > >> Special thanks to @Kurt Young’s team for all the kind help.
> > >>
> > >> Special thanks to @Piotr Nowojski for all the valuable feedback and
> help
> > >> suggestions.
> > >>
> > >>
> > >> Best,
> > >>
> > >> Liya Fan
> > >>
> > >
> >
> > --
> > Best Regards
> >
> > Jeff Zhang
> >
>
>