Hi Xingbo,
+1 from my side
Thanks for the clarification. For your use case the parameter size and
therefore serialization overhead was the limiting factor. I have seen
use cases where that is not the concern, because the Python logic
itself is heavy and dwarfs the protocol overhead (for example when
interacting with external systems from the UDF). Hence it is good to
give users options to optimize for their application requirements.
Cheers,
Thomas
On Tue, Jan 11, 2022 at 3:44 AM Xingbo Huang <hxbks...@gmail.com> wrote:
>
> Hi everyone,
>
> Thanks to all of you for the discussion.
> If there are no objections, I would like to start a vote thread tomorrow.
>
> Best,
> Xingbo
>
> Xingbo Huang <hxbks...@gmail.com> 于2022年1月7日周五 16:18写道：
>
> > Hi Till,
> >
> > I have written a more complicated PyFlink job. Compared with the previous
> > single python udf job, there is an extra stage of converting between table
> > and datastream. Besides, I added a python map function for the job. Because
> > python datastream has not yet implemented Thread mode, the python map
> > function operator is still running in Process Mode.
> >
> > ```
> > source = t_env.from_path("source_table") # schema [id: String, d:int]
> >
> > @udf(result_type=DataTypes.STRING(), func_type="general")
> > def upper(x):
> > return x.upper()
> >
> > t_env.create_temporary_system_function("upper", upper)
> > # python map function
> > ds = t_env.to_data_stream(source) \
> > .map(lambda x: x, output_type=Types.ROW_NAMED(["id", "d"],
> >
> > [Types.STRING(),
> >
> > Types.INT()]))
> >
> > t = t_env.from_data_stream(ds)
> > t.select('upper(id)').execute_insert('sink_table')
> > ```
> >
> > The input data size is 1k.
> >
> > Mode | QPS
> > Process Mode | 3w
> > Thread Mode + Process mode | 4w
> >
> > From the table, we can find that the nodes run in Process Mode is the
> > performance bottleneck of the job.
> >
> > Best,
> > Xingbo
> >
> > Till Rohrmann <trohrm...@apache.org> 于2022年1月5日周三 23:16写道：
> >
> >> Thanks for the detailed answer Xingbo. Quick question on the last figure
> >> in
> >> the FLIP. You said that this is a real world Flink stream SQL job. The
> >> title of the graph says UDF(String Upper). So do I understand correctly
> >> that string upper is the real world use case you have measured? What I
> >> wanted to ask is how a slightly more complex Flink Python job (involving
> >> shuffles, with back pressure, etc.) performs using the thread and process
> >> mode respectively.
> >>
> >> If the mode solely needs changes in the Python part of Flink, then I don't
> >> have any concerns from the runtime perspective.
> >>
> >> Cheers,
> >> Till
> >>
> >> On Wed, Jan 5, 2022 at 1:55 PM Xingbo Huang <hxbks...@gmail.com> wrote:
> >>
> >> > Hi Till and Thomas,
> >> >
> >> > Thanks a lot for joining the discussion.
> >> >
> >> > For Till:
> >> >
> >> > >>> Is the slower performance currently the biggest pain point for our
> >> > Python users? What else are our Python users mainly complaining about?
> >> >
> >> > PyFlink users are most concerned about two parts, one is better
> >> usability,
> >> > the other is performance. Users often make some benchmarks when they
> >> > investigate pyflink[1][2] at the beginning to decide whether to use
> >> > PyFlink. The performance of a PyFlink job depends on two parts, one is
> >> the
> >> > overhead of the PyFlink framework, and the other is the Python function
> >> > complexity implemented by the user. In the Python ecosystem, there are
> >> many
> >> > libraries and tools that can help Python users improve the performance
> >> of
> >> > their custom functions, such as pandas[3], numba[4] and cython[5]. So we
> >> > hope that the framework overhead of PyFlink itself can also be reduced.
> >> >
> >> > >>> Concerning the proposed changes, are there any changes required on
> >> the
> >> > runtime side (changes to Flink)? How will the deployment and memory
> >> > management be affected when using the thread execution mode?
> >> >
> >> > The changes on PyFlink Runtime mentioned here are actually only
> >> > modifications of PyFlink custom Operators, such as
> >> > PythonScalarFunctionOperator[6], which won't affect deployment and
> >> memory
> >> > management.
> >> >
> >> > >>> One more question that came to my mind: How much performance
> >> > improvement dowe gain on a real-world Python use case? Were the
> >> > measurements more like micro benchmarks where the Python UDF was called
> >> w/o
> >> > the overhead of Flink? I would just be curious how much the Python
> >> > component contributes to the overall runtime of a real world job. Do we
> >> > have some data on this?
> >> >
> >> > The last figure I put in FLIP is the performance comparison of three
> >> real
> >> > Flink Stream Sql Jobs. They are a Java UDF job, a Python UDF job in
> >> Process
> >> > Mode, and a Python UDF job in Thread Mode. The calculated value of QPS
> >> is
> >> > the end-to-end Flink job execution result. As shown in the performance
> >> > comparison chart, the performance of Python udf with the same function
> >> can
> >> > often only reach 20% of Java udf, so the performance of python udf will
> >> > often become the performance bottleneck in a PyFlink job.
> >> >
> >> > For Thomas:
> >> >
> >> > The first time that I realized the framework overhead of various IPC
> >> > (socket, grpc, shared memory) cannot be ignored in some scenarios is
> >> due to
> >> > an image algorithm prediction job of PyFlink. Its input parameters are a
> >> > series of huge image binary arrays, and its data size is bigger than 1G.
> >> > The performance overhead of serialization/deserialization has become an
> >> > important part of its poor performance. Although this job is a bit
> >> extreme,
> >> > through measurement, we did find the impact of the
> >> > serialization/deserialization overhead caused by larger size parameters
> >> on
> >> > the performance of the IPC framework.
> >> >
> >> > >>> As I understand it, you measured the difference in throughput for
> >> UPPER
> >> > between process and embedded mode and the difference is 50% increased
> >> > throughput?
> >> >
> >> > This 50% is the result when the data size is less than 100byte. When the
> >> > data size reaches 1k, the performance of the Embedded Mode will reach
> >> about
> >> > 3.5 times the performance of the Process Mode shown in the FLIP. When
> >> the
> >> > data reaches 1M, the performance of Embedded Mode can reach 5 times the
> >> > performance of the Process Mode. The biggest difference here is that in
> >> > Embedded Mode, input/result data does not need to be
> >> > serialized/deserialized.
> >> >
> >> > >>> Is that a typical UDF in your usage?
> >> >
> >> > The reason for choosing UPPER is that a simpler udf implementation can
> >> make
> >> > it easier to evaluate the performance of different execution modes.
> >> >
> >> > >>> What do you observe when the function becomes more complex?
> >> >
> >> > We can analyze the QPS of the framework (process mode or embedded mode)
> >> and
> >> > the QPS of the UDF calculation logic separately. A more complex UDF
> >> means
> >> > that it is a UDF with a smaller QPS. The main factors that affect the
> >> > framework QPS are data type of parameters, number of parameters and
> >> size of
> >> > parameters, which will greatly affect the serialization/deserialization
> >> > overhead in Process Mode.
> >> >
> >> > The purpose of introducing thread mode is not to replace Process mode,
> >> but
> >> > to supplement Python udf usage scenarios such as cep and join, and some
> >> > scenarios where higher performance is pursued. Compared with Thread
> >> mode,
> >> > Process Mode has better isolation, which can solve the limitation of
> >> thread
> >> > mode in some scenarios such as session mode.
> >> >
> >> > [1] https://www.mail-archive.com/user@flink.apache.org/msg42760.html
> >> > [2] https://www.mail-archive.com/user@flink.apache.org/msg44975.html
> >> > [3] https://pandas.pydata.org/
> >> > [4] https://cython.org/
> >> > [5] https://numba.pydata.org/
> >> > [6]
> >> >
> >> >
> >> https://github.com/apache/flink/blob/master/flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/scalar/PythonScalarFunctionOperator.java
> >> >
> >> > Best,
> >> > Xingbo
> >> >
> >> > Thomas Weise <t...@apache.org> 于2022年1月4日周二 04:23写道：
> >> >
> >> > > Interesting discussion. It caught my attention because I was also
> >> > > interested in the Beam fn execution overhead a few years ago.
> >> > >
> >> > > We found back then that while in theory the fn protocol overhead is
> >> > > very significant, for realistic function workloads that overhead was
> >> > > negligible. And of course it all depends on the use case. It might be
> >> > > worthwhile to quantify a couple more scenarios.
> >> > >
> >> > > As I understand it, you measured the difference in throughput for
> >> > > UPPER between process and embedded mode and the difference is 50%
> >> > > increased throughput? Is that a typical UDF in your usage? What do you
> >> > > observe when the function becomes more complex?
> >> > >
> >> > > Thanks,
> >> > > Thomas
> >> > >
> >> > > On Mon, Jan 3, 2022 at 5:52 AM Till Rohrmann <trohrm...@apache.org>
> >> > wrote:
> >> > > >
> >> > > > One more question that came to my mind: How much performance
> >> > improvement
> >> > > do
> >> > > > we gain on a real-world Python use case? Were the measurements more
> >> > like
> >> > > > micro benchmarks where the Python UDF was called w/o the overhead of
> >> > > Flink?
> >> > > > I would just be curious how much the Python component contributes to
> >> > the
> >> > > > overall runtime of a real world job. Do we have some data on this?
> >> > > >
> >> > > > Cheers,
> >> > > > Till
> >> > > >
> >> > > > On Mon, Jan 3, 2022 at 11:45 AM Till Rohrmann <trohrm...@apache.org
> >> >
> >> > > wrote:
> >> > > >
> >> > > > > Hi Xingbo,
> >> > > > >
> >> > > > > Thanks for creating this FLIP. I have two general questions about
> >> the
> >> > > > > motivation for this FLIP because I have only very little exposure
> >> to
> >> > > our
> >> > > > > Python users:
> >> > > > >
> >> > > > > Is the slower performance currently the biggest pain point for our
> >> > > Python
> >> > > > > users?
> >> > > > >
> >> > > > > What else are our Python users mainly complaining about?
> >> > > > >
> >> > > > > Concerning the proposed changes, are there any changes required on
> >> > the
> >> > > > > runtime side (changes to Flink)? How will the deployment and
> >> memory
> >> > > > > management be affected when using the thread execution mode?
> >> > > > >
> >> > > > > Cheers,
> >> > > > > Till
> >> > > > >
> >> > > > > On Fri, Dec 31, 2021 at 9:46 AM Xingbo Huang <hxbks...@gmail.com>
> >> > > wrote:
> >> > > > >
> >> > > > >> Hi Wei,
> >> > > > >>
> >> > > > >> Thanks a lot for your feedback. Very good questions!
> >> > > > >>
> >> > > > >> >>> 1. It seems that we dynamically load an embedded Python and
> >> user
> >> > > > >> dependencies in the TM process. Can they be uninstalled cleanly
> >> > after
> >> > > the
> >> > > > >> task finished? i.e. Can we use the Thread Mode in session mode
> >> and
> >> > > Pyflink
> >> > > > >> shell?
> >> > > > >>
> >> > > > >> I mentioned the limitation of this part in FLIP. There is no
> >> problem
> >> > > > >> without changing the python interpreter, but if you need to
> >> change
> >> > the
> >> > > > >> python interpreter, there is really no way to reload the Python
> >> > > library.
> >> > > > >> The problem is mainly caused by many Python libraries having an
> >> > > assumption
> >> > > > >> that they own the process alone.
> >> > > > >>
> >> > > > >> >>> 2. Does one TM have only one embedded Python running at the
> >> same
> >> > > time?
> >> > > > >> If all the Python operator in the TM share the same PVM, will
> >> there
> >> > > be a
> >> > > > >> loss in performance?
> >> > > > >>
> >> > > > >> Your understanding is correct that one TM have only one embedded
> >> > > Python
> >> > > > >> running at the same time. I guess you are worried about the
> >> > > performance
> >> > > > >> loss of multi threads caused by Python GIL. There is a one-to-one
> >> > > > >> correspondence between Java worker thread and Python
> >> > subinterpreters.
> >> > > > >> Although the subinterpreters has not yet completely overcome the
> >> GIL
> >> > > > >> sharing problem(The Python community’s recent plan for a
> >> > > per-interpreter
> >> > > > >> GIL is also under discussion[1]), the performance of
> >> subinterpreters
> >> > > is
> >> > > > >> very close to that of multiprocessing [2].
> >> > > > >>
> >> > > > >> >>> 3. How do we load the relevant c library if the
> >> > python.executable
> >> > > is
> >> > > > >> provided by users?
> >> > > > >>
> >> > > > >> Once python.executable is provided, PEMJA will dynamically load
> >> the
> >> > > > >> CPython
> >> > > > >> library (libpython.*so or libpython.*dylib) and pemja.so
> >> installed
> >> > in
> >> > > the
> >> > > > >> python environment.
> >> > > > >>
> >> > > > >> >>> May there be a risk of version conflicts?
> >> > > > >>
> >> > > > >> I understand that this question is actually discussing whether
> >> C/C++
> >> > > has a
> >> > > > >> way to solve the problem of relying on different versions of a
> >> > > library.
> >> > > > >> First of all, we know that if there is only static linking, there
> >> > > will be
> >> > > > >> no such problem. And I have studied the source code of
> >> CPython[3],
> >> > > and
> >> > > > >> there is no usage of dynamic linking. The rest is the case where
> >> > > dynamic
> >> > > > >> linking is used in the C library written by the users. There are
> >> > many
> >> > > ways
> >> > > > >> to solve this problem with dynamic linking, but after all, this
> >> > > library is
> >> > > > >> written by users, and it is difficult for us to guarantee that
> >> there
> >> > > will
> >> > > > >> be no conflicts. At this time, Process Mode will be the choice of
> >> > falk
> >> > > > >> back.
> >> > > > >>
> >> > > > >> [1]
> >> > > > >>
> >> > > > >>
> >> > >
> >> >
> >> https://mail.python.org/archives/list/python-...@python.org/thread/S5GZZCEREZLA2PEMTVFBCDM52H4JSENR/#RIK75U3ROEHWZL4VENQSQECB4F4GDELV
> >> > > > >> [2]
> >> > > > >>
> >> > > > >>
> >> > >
> >> >
> >> https://mail.python.org/archives/list/python-...@python.org/thread/PNLBJBNIQDMG2YYGPBCTGOKOAVXRBJWY/#L5OXHXPFONRKLR3W6U46LUSUIBN4FCZQ
> >> > > > >> [3] https://github.com/python/cpython
> >> > > > >>
> >> > > > >> Best,
> >> > > > >> Xingbo
> >> > > > >>
> >> > > > >> Wei Zhong <weizhong0...@gmail.com> 于2021年12月31日周五 11:49写道：
> >> > > > >>
> >> > > > >> > Hi Xingbo,
> >> > > > >> >
> >> > > > >> > Thanks for creating this FLIP. Big +1 for it!
> >> > > > >> >
> >> > > > >> > I have some question about the Thread Mode:
> >> > > > >> >
> >> > > > >> > 1. It seems that we dynamically load an embedded Python and
> >> user
> >> > > > >> > dependencies in the TM process. Can they be uninstalled cleanly
> >> > > after
> >> > > > >> the
> >> > > > >> > task finished? i.e. Can we use the Thread Mode in session mode
> >> and
> >> > > > >> Pyflink
> >> > > > >> > shell?
> >> > > > >> >
> >> > > > >> > 2. Does one TM have only one embedded Python running at the
> >> same
> >> > > time?
> >> > > > >> If
> >> > > > >> > all the Python operator in the TM share the same PVM, will
> >> there
> >> > be
> >> > > a
> >> > > > >> loss
> >> > > > >> > in performance?
> >> > > > >> >
> >> > > > >> > 3. How do we load the relevant c library if the
> >> python.executable
> >> > is
> >> > > > >> > provided by users? May there be a risk of version conflicts?
> >> > > > >> >
> >> > > > >> > Best,
> >> > > > >> > Wei
> >> > > > >> >
> >> > > > >> >
> >> > > > >> > > 2021年12月29日 上午11:56，Xingbo Huang <hxbks...@gmail.com> 写道：
> >> > > > >> > >
> >> > > > >> > > Hi everyone,
> >> > > > >> > >
> >> > > > >> > > I would like to start a discussion thread on "Support PyFlink
> >> > > Runtime
> >> > > > >> > > Execution in Thread Mode"
> >> > > > >> > >
> >> > > > >> > > We have provided PyFlink Runtime framework to support Python
> >> > > > >> user-defined
> >> > > > >> > > functions since Flink 1.10. The PyFlink Runtime framework is
> >> > > called
> >> > > > >> > Process
> >> > > > >> > > Mode, which depends on an inter-process communication
> >> > architecture
> >> > > > >> based
> >> > > > >> > on
> >> > > > >> > > the Apache Beam Portability framework. Although starting a
> >> > > dedicated
> >> > > > >> > > process to execute Python user-defined functions could have
> >> > better
> >> > > > >> > resource
> >> > > > >> > > isolation, it will bring greater resource and performance
> >> > > overhead.
> >> > > > >> > >
> >> > > > >> > > In order to overcome the resource and performance problems on
> >> > > Process
> >> > > > >> > Mode,
> >> > > > >> > > we will propose a new execution mode which executes Python
> >> > > > >> user-defined
> >> > > > >> > > functions in the same thread instead of a separate process.
> >> > > > >> > >
> >> > > > >> > > I have drafted the FLIP-206[1]. Please feel free to reply to
> >> > this
> >> > > > >> email
> >> > > > >> > > thread. Looking forward to your feedback!
> >> > > > >> > >
> >> > > > >> > > Best,
> >> > > > >> > > Xingbo
> >> > > > >> > >
> >> > > > >> > > [1]
> >> > > > >> > >
> >> > > > >> >
> >> > > > >>
> >> > >
> >> >
> >> https://cwiki.apache.org/confluence/display/FLINK/FLIP-206%3A+Support+PyFlink+Runtime+Execution+in+Thread+Mode
> >> > > > >> >
> >> > > > >> >
> >> > > > >>
> >> > > > >
> >> > >
> >> >
> >>
> >
