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 > > >> > > > >> > > > >> > > > >
