pyflink使用的一些疑问

你好，

最近项目想使用flink进行分布式计算，之前项目是Python的pandas项目，想尝试用pyflink进行项目改造，在使用dataset做批处理时，对于Java的版本没有相关map
reduce函数，所以有以下疑问：
1.Python flink的SDK还没支持dataset吗？
2.是不是有其他替代方法？
3.如果还没支持，有计划支持的时间吗？
4.flink table为啥不支持map reduce操作？
5.我们项目使用dataframe来处理数据，能放到flink上做分布式运算吗？dataframe直接转化为table的方式，table不支持map
reduce操作，对应pandas项目改造成flink，有什么好的建议么？
6. datastream api为什么没有实现Windows方法？后面版本会支持吗？

非常感谢，十分看好flink，希望社区越做越大，辛苦了！

您好，
    目前同样在做pyflink 结合pandas的分布式计算调研和尝试，对于您的问题，仅有一些经验性的分享。
    pyflink以后应该都会集成到DataStream，所以应该不会再支持DataSet；
    不建议在计算中间采用 table.to_pandas（）的方式进行table和dataFrame互转，会影响计算效率；
    目前采用的计算效率较好的方式，是定义pandas类型的udf/udaf方式，但相较java版接口同样的方式，pyflink还是会慢很多；
    个人感觉，pyflink耗时较多的地方，还是sql_query的操作，相同sql语句，执行效率上较java差别还是很大的。
    以上仅个人使用感觉，若存在问题，欢迎路过大佬批评指正~
    还有，因为调研相同领域，希望能交流调研新发现，感谢~祝好~




[hidden email]
 
发件人： qian he
发送时间： 2021-03-14 18:59
收件人： user-zh-flink
主题： pyflink使用的一些疑问
你好，
 
最近项目想使用flink进行分布式计算，之前项目是Python的pandas项目，想尝试用pyflink进行项目改造，在使用dataset做批处理时，对于Java的版本没有相关map
reduce函数，所以有以下疑问：
1.Python flink的SDK还没支持dataset吗？
2.是不是有其他替代方法？
3.如果还没支持，有计划支持的时间吗？
4.flink table为啥不支持map reduce操作？
5.我们项目使用dataframe来处理数据，能放到flink上做分布式运算吗？dataframe直接转化为table的方式，table不支持map
reduce操作，对应pandas项目改造成flink，有什么好的建议么？
6. datastream api为什么没有实现Windows方法？后面版本会支持吗？
 
非常感谢，十分看好flink，希望社区越做越大，辛苦了！

[Xingbo Huang]

Hi，

补充回答两点
1. 现在Table上是支持sliding window和Tumpling Window的Pandas UDAF[1]的,
在1.13会支持session
window的UDAF的支持。对于datastream上window的支持，对于上述几种window，你可以转到table上去操作，对于自定义window，datastream会在1.13支持。

2. 关于性能问题，如果你不使用Python
UDFs的话，本质就是跑的Java的代码，python起的作用只是在客户端编译JobGraph的作用，所以不存在说Python
sql_update的运行性能比Java的慢，因为实际运行的代码是一模一样的。对于你使用了Python UDF的话，由于相比Java UDF,
多了IPC的通信开销，以及Python本身的性能就不如Java
Code，目前性能差别大概在6到7倍，我们也一直在性能上做努力，未来希望做到的是完全赶上Java code，甚至C code的性能。

[1]
https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/python/table-api-users-guide/udfs/vectorized_python_udfs.html#vectorized-aggregate-functions

Best,
Xingbo

xiaoyue <[hidden email]> 于2021年3月16日周二 上午11:42写道：

> 您好，
>     目前同样在做pyflink 结合pandas的分布式计算调研和尝试，对于您的问题，仅有一些经验性的分享。
>     pyflink以后应该都会集成到DataStream，所以应该不会再支持DataSet；
>     不建议在计算中间采用 table.to_pandas（）的方式进行table和dataFrame互转，会影响计算效率；
>     目前采用的计算效率较好的方式，是定义pandas类型的udf/udaf方式，但相较java版接口同样的方式，pyflink还是会慢很多；
>     个人感觉，pyflink耗时较多的地方，还是sql_query的操作，相同sql语句，执行效率上较java差别还是很大的。
>     以上仅个人使用感觉，若存在问题，欢迎路过大佬批评指正~
>     还有，因为调研相同领域，希望能交流调研新发现，感谢~祝好~
>
>
>
>
> [hidden email]
>
> 发件人： qian he
> 发送时间： 2021-03-14 18:59
> 收件人： user-zh-flink
> 主题： pyflink使用的一些疑问
> 你好，
>
>
> 最近项目想使用flink进行分布式计算，之前项目是Python的pandas项目，想尝试用pyflink进行项目改造，在使用dataset做批处理时，对于Java的版本没有相关map
> reduce函数，所以有以下疑问：
> 1.Python flink的SDK还没支持dataset吗？
> 2.是不是有其他替代方法？
> 3.如果还没支持，有计划支持的时间吗？
> 4.flink table为啥不支持map reduce操作？
> 5.我们项目使用dataframe来处理数据，能放到flink上做分布式运算吗？dataframe直接转化为table的方式，table不支持map
> reduce操作，对应pandas项目改造成flink，有什么好的建议么？
> 6. datastream api为什么没有实现Windows方法？后面版本会支持吗？
>
> 非常感谢，十分看好flink，希望社区越做越大，辛苦了！
>

[肖越]

Hi, Xingbo
    想跟您了解一下关于sql_query执行上的细节，flink1.12版本底层执行sql语句的过程中，是否有谓词下退的优化？
    从相关的代码测试结果看：
    1. pyflink1.11版本的connector定义支持参数read.query来获取数据，执行效率很高，猜测这部分执行交由数据库完成；
    2. pyflink1.12版本取消了read.query参数，当定义多个数据源执行join等操作时，耗时很明显(pyflink)
    所以，基于上述这种情况，想跟您请教一下这部分耗时，也是因为python的语言缺陷，或者ipc开销？还是底层的实现设计导致的呢？
    感谢~
在 2021-03-16 14:27:22，"Xingbo Huang" <[hidden email]> 写道：

>Hi，
>
>补充回答两点
>1. 现在Table上是支持sliding window和Tumpling Window的Pandas UDAF[1]的,
>在1.13会支持session
>window的UDAF的支持。对于datastream上window的支持，对于上述几种window，你可以转到table上去操作，对于自定义window，datastream会在1.13支持。
>
>2. 关于性能问题，如果你不使用Python
>UDFs的话，本质就是跑的Java的代码，python起的作用只是在客户端编译JobGraph的作用，所以不存在说Python
>sql_update的运行性能比Java的慢，因为实际运行的代码是一模一样的。对于你使用了Python UDF的话，由于相比Java UDF,
>多了IPC的通信开销，以及Python本身的性能就不如Java
>Code，目前性能差别大概在6到7倍，我们也一直在性能上做努力，未来希望做到的是完全赶上Java code，甚至C code的性能。
>
>[1]
>https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/python/table-api-users-guide/udfs/vectorized_python_udfs.html#vectorized-aggregate-functions
>
>Best,
>Xingbo
>
>xiaoyue <[hidden email]> 于2021年3月16日周二 上午11:42写道：
>
>> 您好，
>>     目前同样在做pyflink 结合pandas的分布式计算调研和尝试，对于您的问题，仅有一些经验性的分享。
>>     pyflink以后应该都会集成到DataStream，所以应该不会再支持DataSet；
>>     不建议在计算中间采用 table.to_pandas（）的方式进行table和dataFrame互转，会影响计算效率；
>>     目前采用的计算效率较好的方式，是定义pandas类型的udf/udaf方式，但相较java版接口同样的方式，pyflink还是会慢很多；
>>     个人感觉，pyflink耗时较多的地方，还是sql_query的操作，相同sql语句，执行效率上较java差别还是很大的。
>>     以上仅个人使用感觉，若存在问题，欢迎路过大佬批评指正~
>>     还有，因为调研相同领域，希望能交流调研新发现，感谢~祝好~
>>
>>
>>
>>
>> [hidden email]
>>
>> 发件人： qian he
>> 发送时间： 2021-03-14 18:59
>> 收件人： user-zh-flink
>> 主题： pyflink使用的一些疑问
>> 你好，
>>
>>
>> 最近项目想使用flink进行分布式计算，之前项目是Python的pandas项目，想尝试用pyflink进行项目改造，在使用dataset做批处理时，对于Java的版本没有相关map
>> reduce函数，所以有以下疑问：
>> 1.Python flink的SDK还没支持dataset吗？
>> 2.是不是有其他替代方法？
>> 3.如果还没支持，有计划支持的时间吗？
>> 4.flink table为啥不支持map reduce操作？
>> 5.我们项目使用dataframe来处理数据，能放到flink上做分布式运算吗？dataframe直接转化为table的方式，table不支持map
>> reduce操作，对应pandas项目改造成flink，有什么好的建议么？
>> 6. datastream api为什么没有实现Windows方法？后面版本会支持吗？
>>
>> 非常感谢，十分看好flink，希望社区越做越大，辛苦了！
>>

[Xingbo Huang]

Hi,

其实pyflink作业就两种，一种是用了python udf的，一种是没用python udf
1. 对于没用python
udf的作业，你写的所有python代码就是api层调用，只负责在客户端编译作业。你可以认为实际运行的作业代码全都是java的同一套代码，都是在JVM里面跑的，也就不存在性能差别，如果你觉得哪个操作性能不行，那就得去分析java对应算子的性能问题。
2. 对于用了python
udf的作业，因为你写的udf函数内容是python代码，这种代码在运行时JVM不认识，你需要有PVM执行这种代码，所以会起python进程专门执行udf里面的内容，所以涉及到IPC通信等等。

不知道我描述清楚没有，总的来说，就是你不用python udf的话，那就等同是你写了一个java的flink作业。

Best,
Xingbo

xiaoyue <[hidden email]> 于2021年3月17日周三 下午12:03写道：

> Hi, Xingbo
>     想跟您了解一下关于sql_query执行上的细节，flink1.12版本底层执行sql语句的过程中，是否有谓词下退的优化？
>     从相关的代码测试结果看：
>     1. pyflink1.11版本的connector定义支持参数read.query来获取数据，执行效率很高，猜测这部分执行交由数据库完成；
>     2. pyflink1.12版本取消了read.query参数，当定义多个数据源执行join等操作时，耗时很明显(pyflink)
>     所以，基于上述这种情况，想跟您请教一下这部分耗时，也是因为python的语言缺陷，或者ipc开销？还是底层的实现设计导致的呢？
>     感谢~
> 在 2021-03-16 14:27:22，"Xingbo Huang" <[hidden email]> 写道：
> >Hi，
> >
> >补充回答两点
> >1. 现在Table上是支持sliding window和Tumpling Window的Pandas UDAF[1]的,
> >在1.13会支持session
>
> >window的UDAF的支持。对于datastream上window的支持，对于上述几种window，你可以转到table上去操作，对于自定义window，datastream会在1.13支持。
> >
> >2. 关于性能问题，如果你不使用Python
> >UDFs的话，本质就是跑的Java的代码，python起的作用只是在客户端编译JobGraph的作用，所以不存在说Python
> >sql_update的运行性能比Java的慢，因为实际运行的代码是一模一样的。对于你使用了Python UDF的话，由于相比Java UDF,
> >多了IPC的通信开销，以及Python本身的性能就不如Java
> >Code，目前性能差别大概在6到7倍，我们也一直在性能上做努力，未来希望做到的是完全赶上Java code，甚至C code的性能。
> >
> >[1]
> >
> https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/python/table-api-users-guide/udfs/vectorized_python_udfs.html#vectorized-aggregate-functions
> >
> >Best,
> >Xingbo
> >
> >xiaoyue <[hidden email]> 于2021年3月16日周二 上午11:42写道：
> >
> >> 您好，
> >>     目前同样在做pyflink 结合pandas的分布式计算调研和尝试，对于您的问题，仅有一些经验性的分享。
> >>     pyflink以后应该都会集成到DataStream，所以应该不会再支持DataSet；
> >>     不建议在计算中间采用 table.to_pandas（）的方式进行table和dataFrame互转，会影响计算效率；
> >>     目前采用的计算效率较好的方式，是定义pandas类型的udf/udaf方式，但相较java版接口同样的方式，pyflink还是会慢很多；
> >>     个人感觉，pyflink耗时较多的地方，还是sql_query的操作，相同sql语句，执行效率上较java差别还是很大的。
> >>     以上仅个人使用感觉，若存在问题，欢迎路过大佬批评指正~
> >>     还有，因为调研相同领域，希望能交流调研新发现，感谢~祝好~
> >>
> >>
> >>
> >>
> >> [hidden email]
> >>
> >> 发件人： qian he
> >> 发送时间： 2021-03-14 18:59
> >> 收件人： user-zh-flink
> >> 主题： pyflink使用的一些疑问
> >> 你好，
> >>
> >>
> >>
> 最近项目想使用flink进行分布式计算，之前项目是Python的pandas项目，想尝试用pyflink进行项目改造，在使用dataset做批处理时，对于Java的版本没有相关map
> >> reduce函数，所以有以下疑问：
> >> 1.Python flink的SDK还没支持dataset吗？
> >> 2.是不是有其他替代方法？
> >> 3.如果还没支持，有计划支持的时间吗？
> >> 4.flink table为啥不支持map reduce操作？
> >>
> 5.我们项目使用dataframe来处理数据，能放到flink上做分布式运算吗？dataframe直接转化为table的方式，table不支持map
> >> reduce操作，对应pandas项目改造成flink，有什么好的建议么？
> >> 6. datastream api为什么没有实现Windows方法？后面版本会支持吗？
> >>
> >> 非常感谢，十分看好flink，希望社区越做越大，辛苦了！
> >>
>