Re: Pyflink提交

集群端的 Python 环境中没有安装 PyFlink： ***/python3 这个环境 On Fri, Nov 25, 2022 at 4:02 PM 程龙 <13162790...@163.com> wrote: > 在使用pyflink提交任务时，部署模式onyarn > 1 在不使用Map等算子下如下参数 能够提交成功 并且运行 > .flink run -ynm pytest -m yarn-cluster -pyclientexec ***/python3 > -pyexec ***/python3 -pyarch *** /python3.6.8.zip

Re: pyflink目前map、flatmap都是process实现，那么process当前如何支持sideoutput呢？

是的，当前 PyFlink 还不支持 side output，side output 的支持已经完成开发，会在接下来发布的 1.16 版本中支持。 On Thu, Aug 4, 2022 at 11:50 AM yidan zhao wrote: > 1 需求是根据输入流，根据字段判定，拆分并输出为2个流。 > > 2 目前看 pyflink 的 api，貌似不支持 sideoutput。 > > 3 虽然可以基于输入流 A，连续处理2次，即输入流 A 流向算子 B 和算子 C，分别筛选自己需要的数据进行处理。但这样会导致数据重复传输。 >

Re: pyflink 和 add_jars 的 add_classpaths 路径。

1）尽量用 fat jar，比如 flink-sql-connector-kafka-1.15.0.jar 2）本地运行的话，是mini-cluster方式，执行Flink作业的时候才启动JM/TM，不需要提前部署 Flink。 On Thu, Jul 28, 2022 at 10:08 AM yidan zhao wrote: > 我是本地直接ide内run。 > > Weihua Hu 于2022年7月27日周三 22:10写道： > > > > Hi, 你是怎么提交的任务呢？是提交到远端的 session cluster 上吗？有其他的相关日志吗？ > > > > Best,

Re: sql-client pyexec参数生效疑问

有两个参数指定Python解释器： 1）-pyexec，指定的是作业执行过程中，用来运行Python UDF的Python解释器路径 2）-pyclientexec，指定客户端编译作业的时候，用到的Python解释器路径，具体信息可以看一下： https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/python/dependency_management/#python-interpreter-of-client 可以把这个参数-pyclientexec 也加上试试。 On Tue, Jun 7, 2022

Re: pyflink报错：Java gateway process exited before sending its port number

>> java.lang.NoSuchMethodError: org.apache.flink.util.NetUtils.getAvailablePort()I 你的环境是不是不太干净？可以检查一下 PyFlink 安装目录下(site-packages/pyflink/lib) 的那些 jar 包的版本。 On Mon, May 23, 2022 at 4:22 PM RS wrote: > Hi, > 在Pycharm中，测试Pyflink示例代码，启动运行报错，代码为官方文档中的代码 > 参考官方文档： > https://nightlies.apache.org/fl

Re: Re: Python callback server start failed

eSql("DROP FUNCTION IF EXISTS > "+catalogName+"."+defaultDatabase+"."+functionName).print(); > String createSql = "CREATE FUNCTION IF NOT EXISTS > "+catalogName+"."+defaultDatabase+"."+functionName+" AS '" +className+ &

Re: Re: Python callback server start failed

rk.web.filter.CharacterEncodingFilter.doFilterInternal(CharacterEncodingFilter.java:200) > at > org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107) > at > org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193) > a

Re: Python callback server start failed

发一下完整的日志文件？ On Tue, Apr 19, 2022 at 10:53 AM 799590...@qq.com.INVALID <799590...@qq.com.invalid> wrote: > 软件版本 > > flink-1.13.6 > python3.6.8 > 本地win10也安装了python3.6.8且添加了python的环境变量和成功运行了$ python -m pip install > apache-flink==1.13.6 > standalonesession方式部署的，一个JM 两个TM，3台集群都安装了python3.6.8 且安装了pyf

Re: apache-flink - 在pyflink1.14中聚合函数无法输出的问题

循环发送数据的时候，需要确保数据的时间戳是增长的，否则可能会被作为迟到数据丢弃。 On Mon, Dec 6, 2021 at 2:11 PM su wenwen wrote: > hi,zebing! > > You can go to localhost:8081 and see if it works. Also, data written to > Kafka should be in double quotes. > example: > {"amount": 500, "course_code": "97iscn4g8k","event_time":"2021-12-01

Re: apache-flink - 在pyflink1.14中聚合函数无法输出的问题

可能是 watermark 问题：并发多，测试数据少，有些并发没有数据，导致watermark没有增长。 如果是这个原因的话，有两种解决办法： 1）t_env.get_config().get_configuration().set_string("parallelism.default", "1") 2）t_env.get_config().get_configuration().set_string("table.exec.source.idle-timeout", "5000 ms") On Sat, Dec 4, 2021 at 6:25 PM duanzebing wrot

Re: 哪里可以下载到downloads/setup-pyflink-virtual-env.sh脚本

Flink用的1.14.0，venv.zip中的PyFlink版本是多少？ On Sun, Nov 21, 2021 at 7:59 PM Asahi Lee wrote: > Hi! >    我通过如下命令提交成功了，python的参数需求-D方式传入，-py方式传入不生效： > ./flink-1.14.0/bin/flink > run-application -t yarn-application > -Dyarn.provided.lib.dirs="hdfs://nameservice1/user/flink/flinklib" > -Dyarn.application.

Re: 哪里可以下载到downloads/setup-pyflink-virtual-env.sh脚本

刚注意到你用的YARN application模式，PyFlink 1.14.0才支持YARN application模式，主要是新增了命令行选项“ -pyclientexec” 和配置“python.client.executable”： https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/python/python_config/#python-client-executable 对于你这个作业来说，你需要通过使用1.14.0版本，同时添加命令行选项：-pyclientexec venv.zip/ven

Re: 哪里可以下载到downloads/setup-pyflink-virtual-env.sh脚本

-pyexec 指定的是集群端所用的Python环境，客户端需要编译Flink作业，也会依赖Python环境。可以看一下这个文档: https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/dev/python/dependency_management/#python-interpreter-of-client On Thu, Nov 18, 2021 at 9:00 PM Asahi Lee <978466...@qq.com.invalid> wrote: > Hi ! >    我在java Table api

Re: flink 1.13.2 在 Java/Scala 程序中调用 Python UDF函数，通过yarn-application执行，yarn集群的每台机器都需要安装pyflink?

FYI On Wed, Nov 10, 2021 at 9:31 AM Dian Fu wrote: > 也可以通过以下方式： > - Python libraries [1]: 把PyFlink以及其他相关依赖打包，作为依赖指定 > - Python archieves [2]: 构建Python虚拟环境，并在里面安装PyFlink以及其他依赖，作为依赖指定 > > > 但是上述方式相对于直接在集群上安装来说，提交作业的时候，Flink内部需要把相关文件分发到集群节点上，如果文件比较大，有一点的overhead，会降低启动速

Re: flink 1.13.2 在 Java/Scala 程序中调用 Python UDF函数，通过yarn-application执行，yarn集群的每台机器都需要安装pyflink?

也可以通过以下方式： - Python libraries [1]: 把PyFlink以及其他相关依赖打包，作为依赖指定 - Python archieves [2]: 构建Python虚拟环境，并在里面安装PyFlink以及其他依赖，作为依赖指定 但是上述方式相对于直接在集群上安装来说，提交作业的时候，Flink内部需要把相关文件分发到集群节点上，如果文件比较大，有一点的overhead，会降低启动速度。 [1] https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/python/dependency_m

Re: pyflink 1.14.0 udf 执行报错，根据官网写的代码

图挂了，邮件列表不能直接发图片。可以发一下更详细的日志信息吗？ On Tue, Oct 19, 2021 at 6:34 PM xuzh wrote: > 错误日志 > Exception in thread Thread-14: > Traceback (most recent call last): >   File "D:\Anaconda3\envs\py37\lib\threading.py", line 926, in > _bootstrap_inner >     self.run() >   File > "D:\Anaconda3\envs\py37\lib\sit

Re: pyflink 1.14.0 udf 执行报错，根据官网写的代码

我试了一下是可以运行的，可以发一下报错吗？ On Mon, Oct 18, 2021 at 6:44 PM xuzh wrote: > from pyflink.table import ScalarFunction, EnvironmentSettings, > TableEnvironment, DataTypes > from pyflink.table.udf import udf > from pyflink.table.expressions import call, row > > > class HashCode(ScalarFunction): > def

Re: Pyflink中使用ConnectedStream时，key_by后分区的问题

Hi， 是否发一下可复现的完整示例？ Regards, Dian > 2021年7月10日 下午7:44，赵飞 写道： > > 各位好，请教一个问题。 > > 最近我在使用pyflink开发一个模块，主要的功能是基于规则对用户数据进行计算和判断。涉及到两个流：数据流（data）和规则流（rule），两者都包含一个产品id值，所以将该值作为key来分区，处理的代码大致如下： > > --- > results = data.connect(rules).key_by('product_id', > 'product_id').process(MyFunction()) >

Re: pyflink kafka connector报错ByteArrayDeserializer

要用fat jar： https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-kafka_2.11/1.13.1/flink-sql-connector-kafka_2.11-1.13.1.jar > 2021年6月2日 下午2:43，qianhuan <819687...@qq.com> 写道： > > 版本： > python 3.8 > apache-flink 1.13.1 > apache-flink-libraries 1.13.1 > > 代码： > fro

Re: Pyflink jdbc相关

这样试试，把”\”改成”/“： file:///D:/Pyproject/flink-connector-jdbc_2.11-1.13.1.jar > 2021年6月1日 下午5:40，琴师 <1129656...@qq.com> 写道： > > 再请叫一个问题，我在pycharm使用时候引用windows的地址不能引用，比如file:///D:\Pyproject\flink-connector-jdbc_2.11-1.13.1.jar; > 这样不能引入，大佬有用用过吗？ > > > -- 原始邮件 --

Re: Pyflink jdbc相关

Hi， 本地执行： 1）用flink-connector-jdbc_2.11-1.13.1.jar试试？因为PyFlink里默认待的JAR包是scala 2.11的 flink run: 1) 你注册的sink表的名字为“print”，不是”table_sink”，但是在SQL语句里用的table_sink。 > 2021年6月1日 下午4:33，琴师 <1129656...@qq.com> 写道： > > Hi， >    我按着微信分享https://mp.weixin.qq.com/s/GyFTjQl6ch8jc733mpCP7Q > 试着使用pyflink，遇到了问题，

Re: pyflink双流join

用Table API的话，可以看一下这个： https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/dev/table/tableapi/#joins 另外，也可以直接在PyFlink中调用SQL语句： https://ci.apache.org/projects/flink/flink-docs-release-1.13/d

PyFlink调查问卷

Hi all, 【填问卷，抽奖送T恤！】它来了它来了 ~ 为了更好地服务 PyFlink 用户，帮助 PyFlink 用户将 PyFlink 应用到生产环境中，Apache Flink 中文社区接下来计划推出一系列 PyFlink 的相关的文档及参考资料，让 PyFlink 用户得到更多优质的 PyFlink 学习资料！ 为此我们推出这个调查问卷，了解大家感兴趣的内容，希望大家积极参与这个问卷，帮助我们更好的去整理 PyFlink 相关学习资料~ PS：填完问卷后即可参与抽奖，Flink 定制款 Polo 衫送送送！4月30日中午12:00准时开奖哦 ~ https://surv

Re: pyflink kafka connector报错

把flink-connector-kafka_2.11-1.12.2.jar删了 > 2021年4月19日 下午6:08，qianhuan <819687...@qq.com> 写道： > > 感谢回复🙏 > 导入了flink-sql-connector-kafka_2.11-1.12.2.jar和flink-connector-kafka_2.11-1.12.2.jar，并且放到了site-packages/pyflink/lib目录下，还是一样的报错。 > test_source_table_1这个kafka的表应该是创建成功了，是查询的问题吗？ > > > > -- > Se

Re: pyflink kafka connector报错

Hi, 需要使用 fat jar [1]，可以看一下Kafka Table & SQL connector 的文档 [2]. [1] https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-kafka_2.11/1.12.2/flink-sql-connector-kafka_2.11-1.12.2.jar

Re: Re: pyflink 运行提示：Function class 'class org.apache.flink.table.functions.python.PythonScalarFunction' is not serializable

命令：/opt/flink-1.11.2/bin/flink run -m yarn-cluster -ynm traffic -ys 1 > -ytm 1024m -p 1 -py traffic.py > > > > > > -- 原始邮件 -- > > 发 件 人："Dian Fu" > > 发送时间：2021-04-14 23:11:57 > > 收 件 人：user-zh > > 抄 送： > > 主 题：Re

Re: pyflink 运行提示：Function class 'class org.apache.flink.table.functions.python.PythonScalarFunction' is not serializable

你JDK版本多少？ 看起来像是Java环境的问题。这里有一个相似的问题[1]，看下是否有帮助。 [1] https://stackoverflow.com/questions/41265266/how-to-solve-inaccessibleobjectexception-unable-to-make-member-accessible-m On Wed, Apr 14, 2021 at 4:58 PM magichuang wrote: > flink版本：1.11.2 Python版本 3.6 apache-flink==1.11.2, 用的是flink on >

Re: pyflink资源优化问题，请教

处理逻辑看起来应该是没有问题的。 1）可以详细说一下，你说的数据延迟问题吗？现在的qps可以达到多少，预期是多少？ 2）你现在用的哪种部署模式？ 3）并发度的设置可以参考：https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/python/table-api-users-guide/table_environment.html#configuration

Re: flink1.12 Standalone模式发送python脚本任务报错： java.lang.ClassNotFoundException: org.apache.flink.connector.jdbc.table.JdbcRowDataInputFormat

可以看一下： https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/python/table-api-users-guide/dependency_management.html#java-dependency-in-python-program flink-connector-jdbc_2.11-1.12.0.jar和mysql-connector-java-8.0.12.jar，需要放到PyFlink可以找到的地方。 On Mon, Mar 22, 2021 at 1:43 PM xiaoyue <18242

Re: 关于statement输出结果疑问

流作业本身是不支持lazy模式的。 > 2021年3月17日 下午3:33，Shuai Xia 写道： > > Hi,大佬,想问下如果使用Lazy调度模式,情况会是什么样子 > > ------ > 发件人：Dian Fu mailto:dian0511...@gmail.com>> > 发送时间：2021年3月15日(星期一) 15:49 > 收件人：刘杰鸿 mailto:wushang.

Re: Pyflink dataset没有支持相关map reduce函数

Hi, 有几个疑问： 1）你说的map reduce函数具体指的什么？可以举一个例子吗？ 2）DataSet API指的是Java的DataSet API吗？另外，Java的DataSet API会逐步废弃，统一到DataStream API上来，所以PyFlink里不会支持DataSet API，只支持Python Table API和Python DataStream API > 2021年3月13日 上午10:54，nova.he 写道： > > 你好， >       > 最近项目想使用flink进行分布式计算，之前项目是Python的pandas项目，想尝试用pyf

Re: 关于statement输出结果疑问

sink。 > 2021年3月12日 下午10:52，刘杰鸿 写道： > > 我的理解是代码会先把source里面的数据全部插入到sink1，这时候就是1，2。然后执行到add_insert_sql代码的时候，将source数据全部插入到sink2里面，这时候也是1，2。 > 所以输出应该是1，2，1，2 > > -- 原始邮件 ------ > 发件人: "Dian Fu" ; > 发送时间: 2021年3月12日(星期五) 晚上10:24 > 收件人: &q

Re: 关于statement输出结果疑问

可以说一下为什么你觉得输出结果应该是1，2，1，2吗？ 个人觉得现在的输出结果应该没有问题。比如第1条数据，先发送到sink1，再发送到sink2，所以打印了2个1；然后处理第二条数据，打印了2个2 On Mon, Mar 8, 2021 at 9:42 AM 刘杰鸿 wrote: > from pyflink.table import EnvironmentSettings, StreamTableEnvironment > > env_settings = > EnvironmentSettings.new_instance().in_streaming_mode().use_b

Re: Pyflink 提交到本地集群报错

从报错看，似乎是作业运行的时候，找不到pyflink，如果确实是这样的话，有几个解决方案： - 通过API指定集群端的Python路径： set_python_executable，参考 [1] - 通过配置python.executable，参考[2] [1] https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/python/table-api-users-guide/dependency_management.html [2] https://ci.apache.org/projects/flink/flin

Re: Pyflink如何对接HBase？

1）在PyFlink Table API中可以使用所有SQL中支持的connector，所以HBase connector也自然支持，具体使用方式可以看一下文档： https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/python/table-api-users-guide/python_table_api_connectors.html#how-to-use-connectors 2）HBase connector的使用方式可以看一下： https://ci.apache.org/projects/flink/fl

Re: PyFlink UDTF 运行一段时间后报 NullPointerException

这个问题应该和这个JIRA有关系：https://issues.apache.org/jira/browse/FLINK-21434 目前已经在master和1.12.3上修复了，但是1.12.3还没有release。你要不cherry-pick这个fix，自己build一个版本，验证一下？ On Tue, Mar 9, 2021 at 5:27 PM 梁丁文 wrote: > 测试PyFlink UDTF，运行数秒后异常退出。下面是UDTF函数 > > > class Mac(TableFunction): >     def eval(self, body_data): >  

Re: 关于pyflink LATERAL TABLE 问题请教

用的PyFlink版本是多少？另外，如果方便的话，可以提供一个比较容易复现的例子吗？ On Fri, Mar 12, 2021 at 4:57 PM 陈康 <844256...@qq.com> wrote: > 定制UDTF想要拆分字符串、但是报错 java.lang.IndexOutOfBoundsException: Index: 7, Size: 7 > 毫无头绪、有大佬遇到过吗？谢谢！ > > class myKerasMLP(ScalarFunction): > > def eval(self, *args): > ... > # 返回预

Re: pyFlink UDF Caused by: java.lang.RuntimeException: Failed to create stage bundle factory! INFO:root:Initializing python harness:

这个问题应该有人问过，你搜搜看。 另外，如果GC频繁的话，把内存调大看看～ > 在 2021年2月8日，下午5:14，陈康 <844256...@qq.com> 写道： > > 感谢回复...切换了版本...运行报错如下图 。。 > [hadoop@hadoop01 bin]$ pip list | grep flink > apache-flink 1.11.1 > > Caused by: java.util.concurrent.TimeoutException: Heartbeat of TaskManager > with id 397

Re: pyFlink UDF Caused by: java.lang.RuntimeException: Failed to create stage bundle factory! INFO:root:Initializing python harness:

看起来似乎是因为Flink集群的版本和PyFlink的版本不一致导致的：集群装的Flink是1.11.1，PyFlink是1.12.0？ 先把版本都统一一下，再试试。 > 在 2021年2月8日，上午10:28，陈康 <844256...@qq.com> 写道： > > 请教大佬们： 一个最简单pyflink UDF跑起来，报 Failed to create stage bundle factory! > INFO:root:Initializing python harness: 在IdeaIJ上可以运行、大家有遇到过吗？谢谢~ > > /opt/module/flink-1.1

Re: PyFlink Could not read the user code wrapper:org.apache.flink.connector.jdbc.table.JdbcRowDataInputFormat

Hi, flink-connector-jdbc_2.11-1.11.1.jar 有添加在flink/lib下，只能保证在作业执行的时候，可以找到对应的class，在客户端提交的时候，会编译作业，从报错看，是客户端编译作业的时候找不到对应的class。 可以试试这里的方法：https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/python/table-api-users-guide/dependency_management.html#java-dependency > 在 2021年2月7日，上午10:50

Re: PyFlink How to set timeout for UDF

先理解一下你的需求：是说Python UDF的实现，处理一条数据的时间可能非常长，如果很长时间没有执行完，希望作业停止？ > 在 2021年2月3日，下午1:04，苗红宾 写道： > > Hi： > > Hope you are doing well! > > My UDF always running in a long time, so I'm wondering, how to set timeout > for UDF in Pyflink, in order to auto-stop the execution when it running in a > long

Re: HELP!!!! PyFlink UDF加载Keras模型，并注册函数出现的问题

可以发一下你的__init__方法吗？应该是在__init__方法里有不能pickle的对象。 > 在 2021年2月3日，下午6:01，陈康 <844256...@qq.com> 写道： > > > https://blog.csdn.net/weixin_44904816/article/details/108744530 > 看到一篇博客说：“PyFlink以后还可以支持 Tensorflow、Keras”.好吧.. > > > > -- >

Re: [ANNOUNCE] Apache Flink 1.10.3 released

Thanks Xintong for driving this release! Regards, Dian > 在 2021年1月29日，下午5:24，Till Rohrmann 写道： > > Thanks Xintong for being our release manager. Well done! > > Cheers, > Till > > On Fri, Jan 29, 2021 at 9:50 AM Yang Wang > wrote: > Thanks Xintong for driving thi

Re: flink-1.11.1 setMinPauseBetweenCheckpoints不生效

应该是个已知问题，在1.11.2里已经修复了：https://issues.apache.org/jira/browse/FLINK-18856 > 在 2020年12月24日，下午9:34，赵一旦 写道： > > I don't believe what you say... > > nicygan 于2020年12月24日周四 下午7:25写道： > >> dear all: >>我在checkpoint设置中，设置了 >> >> checkpointConfig.setMinPauseBetweenCheckpoints(180_000L) >>

Re: pyflink1.12 使用connector read.query参数报错

'table-name' = 'TS_PF_SEC_YLDRATE' 这一行后面少个逗号 > 在 2020年12月24日，下午2:02，肖越 <18242988...@163.com> 写道： > > 使用DDL 定义connector连接Mysql数据库，想通过发送sql的方式直接获取数据: > source_ddl = """ > CREATE TABLE source_table( >yldrate DECIMAL, >pf_id VARCHAR, >symbol_id VARCHAR)

Re: pyflink 有没有方便的print方式？例如java api中的 .print() ?

可以collect到client端[1]，或者可以看看另外几种方式[2]： [1] https://ci.apache.org/projects/flink/flink-docs-release-1.12/api/python/pyflink.table.html#pyflink.table.TableResult.collect [

Re: Pyflink1.12尝试连接oracle数据，报错findAndCreateTableSource failed

onstructorCommand.java:80) > > at > org.apache.flink.api.python.shaded.py4j.commands.ConstructorCommand.execute(ConstructorCommand.java:69) > > at > org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238) > >

Re: 求教：pyflink的sink是否支持redis connector?

nector-redis > 可以打包成jar包嘛，然后在pyflink里用 > > 对java不熟悉，我看这个页面里面只是对java和scala说了如何用 > > > > > > > > Best， > > MagicHuang > > > > >> -- 原始邮件 -- >> 发 件 人："Dian Fu" >> 发送时间：2020-12-1

Re: Pyflink1.12尝试连接oracle数据，报错findAndCreateTableSource failed

t; > > > > > > > > 在 2020-12-17 10:44:56，"Dian Fu" 写道： >> 1）如Leonard Xu老师所说，目前JDBC connector还不支持oracle >> 2）从报错看，你这还都没有走到那一步就报错了，可以检查一下这些： >>a. JDBC connector的使用方式参考[1]，比如'connector' = 'jdbc'，而不是'connec

Re: 求教：pyflink的sink是否支持redis connector?

t; > > > > > > > > 在 2020-12-17 10:16:13，"Dian Fu" 写道： >> 感谢Xingbo的回复，稍微补充一点：所有Table & SQL支持的connector都可以用在PyFlink中。 >> >> redis的connector没有直接在Flink的代码库里提供，这里有一个，应该也可以用：https://github.com/apache/bahir-flink/tree/master/flink-connector-re

Re: Pyflink1.12尝试连接oracle数据，报错findAndCreateTableSource failed

1）如Leonard Xu老师所说，目前JDBC connector还不支持oracle 2）从报错看，你这还都没有走到那一步就报错了，可以检查一下这些： a. JDBC connector的使用方式参考[1]，比如'connector' = 'jdbc'，而不是'connector.type' = 'jdbc'，这个是老的使用方式 b. JDBC connector的jar需要显式添加相关依赖，可以如何在PyFlink中添加jar依赖，可以看一下[2]。其中JDBC需要的的jar，可以参考[1] [1] https://ci.apache.org/project

Re: 求教：pyflink的sink是否支持redis connector?

感谢Xingbo的回复，稍微补充一点：所有Table & SQL支持的connector都可以用在PyFlink中。 redis的connector没有直接在Flink的代码库里提供，这里有一个，应该也可以用：https://github.com/apache/bahir-flink/tree/master/flink-connector-redis 关于如何在PyFlink中使用connector，可以参考文档：https://ci.ap

Re: pyflink 1.11 运行pyflink作业时报错

看起来是的，找不到JAVA_HOME，显式export一下JAVA_HOME试试？ > 在 2020年11月13日，下午5:06，whh_960101 写道： > > Hi，各位大佬，pyflink 1.11 运行pyflink作业时报错pyflink/pyfink_gateway_server.py 193行 > lauch_gateway_server_process()217行 return Popen（）FileNotFoundError： [Error 2] > No such file or directory: 'java' : 'java'感觉像找不到ja

Re: 提交pyflink作业到YARN集群报错

看一下目录site-packages/pyflink/opt，里面是否有名为flink-python的jar > 在 2020年10月26日，下午4:38，whh_960101 写道： > > Hi，各位大佬， 想请教一下，我使用flink run -m yarn-cluster -p 4 -py > myjob.py,报错java.lang.RuntimeException: Found 0 flink-python jar. > at > org.apache.flink.client.program.PackagedProgram.getJobJarAndDepend

Re: pyflink1.11.0 报错JobExecutionException: Job execution failed.

PyFlink里有Row类型的类, pyflink.table.Row > 在 2020年10月21日，上午9:05，whh_960101 写道： > > Row类型的对象在python中是怎么表示的，字典？ > > > > > > > 在 2020-10-20 20:35:22，"Dian Fu" 写道： >> 你这两个UDF（error_get和headers_get）实际的返回值类型都是string，但是却标成Row类型。如果确实需要返回Row类型，udf的返回值需要返

Re: pyflink1.11.0 报错JobExecutionException: Job execution failed.

你这两个UDF（error_get和headers_get）实际的返回值类型都是string，但是却标成Row类型。如果确实需要返回Row类型，udf的返回值需要返回一个Row类型的对象。 > 在 2020年10月20日，下午7:56，Dian Fu 写道： > > 错误堆栈看着似乎不太完整，有更完整的堆栈吗？ > >> 在 2020年10月20日，下午7:38，whh_960101 写道： >> >> Hi，各位大佬，我处理kafka source Table，print 一个嵌套的json结

Re: pyflink1.11.0 报错JobExecutionException: Job execution failed.

错误堆栈看着似乎不太完整，有更完整的堆栈吗？ > 在 2020年10月20日，下午7:38，whh_960101 写道： > > Hi，各位大佬，我处理kafka source Table，print 一个嵌套的json结果，报错JobExecutionException: Job > execution failed. 这个问题该怎么解决，代码和报错信息如下：@udf(input_types=DataTypes.STRING(), > result_type=DataTypes.BOOLEAN()) >def error_exist(message): >

Re: pyflink sql中select,where都带udf，其中一个udf失效

lan == > > +- PythonCalc(select=[message, kubernetes, clusterName, > error_exist(message) AS f0]) > #感觉应该是这个地方出问题了，这里应该不是select，应该是where或者filter，上面已经有了LogicalFilter(condition=[error_exist($1)]) > > > > > == Physical Execution Plan == > > > > > Stage 3 : O

Re: pyflink sql中select,where都带udf，其中一个udf失效

可以把plan打印出来看一下？打印plan可以参考这个：https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/python/table-api-users-guide/intro_to_table_api.html#explain-tables > 在 2020年10月15日，下午7:02，whh_960101 写道： > > hi， > 我刚才改了一下你的例子[1]，通过from_elements构建一个source表 > 然后使用我的udf > source.where(udf1(msg)=True).se

Re: pyflink如何提交作业到CDP集群中运行

PyFlink支持提交到Standalone／Yarn／K8s集群运行。 Standalone和Yarn： https://ci.apache.org/projects/flink/flink-docs-release-1.11/ops/cli.html K8s: https://issues.apache.org/jira/browse/FLINK-17480?focusedCommentId=17187596&page=com

Re: 问题跟踪

因为你需要根据不同'data'的数据，构造不同的作业逻辑，就必然需要执行作业，获取'data'的值，所以你目前的做法是对的。 1.12发布之后，还可以通过以下方式： 1）table.limit(1).to_pandas：可以只取表中的第一条数据 2）table.limit(1).collect()： limit以及collect 1.12会支持。 > 在 2020年9月9日，下午5:28，whh_960101 写道： > > 回答：对的，我的需求是根据'data'的值，再构造不同的作业逻辑，如何实现，求解答！谢谢！

Re: pyflink execute_insert问题求解答

针对问题1: 你的需求是这样的吗：先获取表中字段'data'的值（第一行的值），根据'data'的值，再构造不同的作业逻辑？ 针对问题2：现在join不支持两个表的字段名重复，可以看一下JIRA [1]，所以目前必须保证两个表的字段名不重复。 [1] https://issues.apache.org/jira/browse/FLINK-18679 > 在 2020年9月9日，下午4:27，whh_960101 写道： > > 问题1： > 我已经生成了一个Table对象main

Re: pyflink execute_insert问题求解答

这两个看起来是同一个问题，1.11是支持的，可以看一下TableEnvironment.create_statement_set(): https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/python/table-api-users-guide/table_environment.html#executeexplain-jobs > 在 2020年9月9日，上午11:31，whh_960101 写道： > > 您好，我使用pyflink时的代码如下，有如下两个问题： > 1. > source = st_

Re: pyflink1.11.1连接hive问题

Invoker.java:244) > > at > org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357) > > at org.apache.flink.api.python.shaded.py4j.Gateway.invoke(Gateway.java:282) > > at > org.apache.flink.api.python.shaded.py4j.commands.AbstractCommand.

Re: pyflink1.11.1连接hive问题

有更完整的log吗？ > 在 2020年9月3日，下午4:12，程龙 <13162790...@163.com> 写道： > > 使用hivecatalog连接 抱一下错误 > > > > > flink py4j.protocol.Py4JJavaError: An error occurred while calling > o10.registerCatalog. : java.lang.NullPointerException

Re: [ANNOUNCE] Apache Flink 1.10.2 released

Thanks ZhuZhu for managing this release and everyone else who contributed to this release! Regards, Dian > 在 2020年8月25日，下午2:22，Till Rohrmann 写道： > > Great news. Thanks a lot for being our release manager Zhu Zhu and to all > others who have contributed to the release! > > Cheers, > Till > >

Re: 【PyFlink】对于数据以Csv()格式写入kafka报错，以及使用python udf时无法启动udf

flink 1.11.0的话，依赖的beam版本是2.19.0，所以先不要用最新的beam 2.23.0 针对你这个问题，应该是在启动过程中，由于某种原因，Python进程启动失败了，常见的原因有：依赖缺失，Python版本不对等等。 按说在log文件里（你发的log信息所在的log文件），应该能看到详细的原因，你的log文件里面没有详细的失败原因吗？ > 在 2020年8月7日，下午1:44，lgs <9925...@qq.com> 写道： > > Hi Jincheng, > > 我现在碰到同样的问题，udf运行的时候会打印这样的log： > 2020-08-07 03:06:4

Re: [DISCUSS] FLIP-133: Rework PyFlink Documentation

Hi Jincheng, Thanks a lot for bringing up this discussion and the proposal. +1 to improve the Python API doc. I have received many feedbacks from PyFlink beginners about the PyFlink doc, e.g. the materials are too few, the Python doc is mixed with the Java doc and it's not easy to find the doc

[ANNOUNCE] Apache Flink 1.11.1 released

The Apache Flink community is very happy to announce the release of Apache Flink 1.11.1, which is the first bugfix release for the Apache Flink 1.11 series. Apache Flink® is an open-source stream processing framework for distributed, high-performing, always-available, and accurate data streamin

Re: pyflink udf中发送rest api会导致udf被调用两次

大部分情况下，可以work，但是有一些边界的情况，可能会有问题。比如第一个sink的作业，由于某种原因，处理得比较慢，延迟比较大？ 也就是说，通常情况下可能没有问题，但是由于这2个作业之间没有任何依赖关系，这个先后顺序是得不到保证的。 我觉得你可以测一下，如果能接受那些极端情况，就可以。 > 在 2020年7月10日，下午5:08，lgs <9925...@qq.com> 写道： > > 代码顺序是指我先写第一个sink的代码，再写第二个sink的代码。 > > 我设置了'connector.write.flush.max-rows' = '1' > 第一个sink没有窗口，所以直

Re: pyflink udf中发送rest api会导致udf被调用两次

我不太明白你说的“代码顺序”指的什么？ 据我所知，应该没有什么太好的办法。从执行图上来看，这2个之间没有依赖关系，所以也就无法保证先后顺序。 如果必须这样干的话，你得从业务的角度想一下，改造一下业务逻辑。 > 在 2020年7月10日，下午4:10，lgs <9925...@qq.com> 写道： > > 这次可以了。谢谢 > > 另外还有一个问题请教一下： > 我实际上是有另一个sink，source是同一个。 > 第一个sink是直接保存kafka数据到DB。 > 第二个sink是读取kafka，tumble window，然后在udf里面去读取DB。 > > 要怎么样保证

Re: pyflink udf中发送rest api会导致udf被调用两次

这样再试试？ tmp_table = st_env.scan("source") \ .where("action === 'Insert'") \ .window(Tumble.over("1.hour").on("actionTime").alias("hourlywindow")) \ .group_by("hourlywindow") \ .select("action.max as action1, conv_string(eventTime.collect) as etlist, hourlywindow.sta

Re: pyflink udf中发送rest api会导致udf被调用两次

好的，针对你这个case，这个是个已知问题：https://issues.apache.org/jira/browse/FLINK-15973 ，暂时还没有修复。 你可以这样改写一下，应该可以绕过去这个问题： table = st_env.scan("source") \ .where("action === 'Insert'") \ .window(Tumble.over("1.hour").on("actionTime").alias("hour

Re: pyflink udf中发送rest api会导致udf被调用两次

Table API的作业在执行之前会经过一系列的rule优化，最终的执行计划，存在一个UDF调用多次的可能，你可以把执行计划打印出来看看(TableEnvironment#explain)。 具体原因，需要看一下作业逻辑。可以发一下你的作业吗？可重现代码即可。 > 在 2020年7月9日，下午5:50，lgs <9925...@qq.com> 写道： > > Hi， > > 我观察到一个现象：我定义了一个tumble window，调用一个python udf，在这个udf里面使用requests发送rest api。 > log显示这个udf会被调用两次。相隔不到一秒。这个是什么

Re: [ANNOUNCE] Yu Li is now part of the Flink PMC

Congrats Yu! Regards, Dian > 在 2020年6月17日，上午10:35，Jark Wu 写道： > > Congratulations Yu! Well deserved! > > Best, > Jark > > On Wed, 17 Jun 2020 at 10:18, Haibo Sun wrote: > >> Congratulations Yu! >> >> Best, >> Haibo >> >> >> At 2020-06-17 09:15:02, "jincheng sun" wrote: >>> Hi all, >>>

Re: pyflink连接elasticsearch5.4问题

gt;> .index("taxiid-cnts") > >> .document_type('taxiidcnt') > >> .key_delimiter("$")) \ > > > > > > 在 2020-06-16 15:38:28，"Dian Fu" 写道： > >I guess it's because the ES

Re: pyflink连接elasticsearch5.4问题

I guess it's because the ES version specified in the job is `6`, however, the jar used is `5`. > 在 2020年6月16日，下午1:47，jack 写道： > > 我这边使用的是pyflink连接es的一个例子，我这边使用的es为5.4.1的版本，pyflink为1.10.1，连接jar包我使用的是 > flink-sql-connector-elasticsearch5_2.11-1.10.1.jar，kafka，json的连接包也下载了，连接kafka测试成功了。 > 连接es的时候

Re: pyflink Table Api 使用udf函数报错 Given parameters do not match any signature.

你传的第二个参数是string，这样试一下？ select("drop_fields(message, array('x'))") 不太确定我是否理解了你的问题（如果上面不行的话，建议你发一下exception） > 在 2020年6月1日，下午1:59，jack 写道： > > > > > > > > 是的，对应参数没有填写正确，感谢； > 另外请教，udf函数是不是目前不支持python的可变参数，我使用可变参数依然会报错参数不对的问题。 > >

Re: pyflink Table Api 使用udf函数报错 Given parameters do not match any signature.

The input types should be as following: input_types=[DataTypes.STRING(), DataTypes.ARRAY(DataTypes.STRING())] Regards, Dian > 在 2020年6月1日，上午10:49，刘亚坤 写道： > > 目前在学习使用pyflink的Table api，请教一个pyflink udf函数报错问题： > > @udf(input_types=[DataTypes.STRING()], result_type=DataTypes.STRING()) > def drop_f

Re: [ANNOUNCE] Apache Flink 1.10.1 released

Thanks Yu for managing this release and everyone else who made this release possible. Good work! Regards, Dian > 在 2020年5月15日，下午6:26，Till Rohrmann 写道： > > Thanks Yu for being our release manager and everyone else who made the > release possible! > > Cheers, > Till > > On Fri, May 15, 2020 a

[ANNOUNCE] Apache Flink 1.9.3 released

Hi everyone, The Apache Flink community is very happy to announce the release of Apache Flink 1.9.3, which is the third bugfix release for the Apache Flink 1.9 series. Apache Flink® is an open-source stream processing framework for distributed, high-performing, always-available, and accurate data

Re: 订阅

需要发邮件到：user-zh-subscr...@flink.apache.org <mailto:user-zh-subscr...@flink.apache.org>来订阅user-zh > 在 2020年4月16日，下午5:30，Dian Fu 写道： > > 需要发邮件到：user-zh-subscr...@flink.apache.org来订阅user-zh > >> 在 2020年4月16日，下午5:28，李朋 <1134415...@qq.com> 写道： >> >> 订阅 >

Re: 订阅

需要发邮件到：user-zh-subscr...@flink.apache.org来订阅user-zh > 在 2020年4月16日，下午5:28，李朋 <1134415...@qq.com> 写道： > > 订阅

Re: flink cep 匹配一段时间类A，B，C事件发生

类似于这样？ AA follow by BB follow by CC AA定义成A or B or C BB定义成（A or B or C）and BB.type != AA.type CC定义成（A or B or C）and CC.type != AA.type and CC.type != BB.type > 在 2020年4月16日，上午8:40，Peihui He 写道： > > hello，all > >我这个边需要匹配一段时间内A，B，C事件同时发生，但是不要求A，B，C事件的顺序，flink cep有什么好的方式不？ > > 有

Re: 关于pyflink连接rabbitmq

PyFlink目前只支持Python Table API，rabbitmq目前还没有提供Table/SQL的connector，如果想在PyFlink里使用rabbitmq，有以下几种解决方案： 1）实现Java的rabbitmq的TableSource/TableSink，可以参考Kafka等connector的实现，基本只需要在现有实现的基础上包装一下即可。 2）在PyFlink作业里使用rabbitmq的source/sink。目前在PyFlink里注册TableSource/TableSink有2种方式： 2.1）如果使用TableEnvironment.register_tab

Re: 关于kafka connector通过python链接

; https://enjoyment.cool/2019/08/28/Apache%20Flink%20%E8%AF%B4%E9%81%93%E7%B3%BB%E5%88%97-%20Python%20API%20%E4%B8%AD%E5%A6%82%E4%BD%95%E4%BD%BF%E7%94%A8%20Kafka/ > > <https://enjoyment.cool/2019/08/28/Apache%20Flink%20%E8%AF%B4%E9%81%93%E7%B3%BB%E5%88%97-%20Python%20API%20%E4%B8%AD%E

Re: 关于kafka connector通过python链接

你指的是Python Table API中如何使用kafka connector的例子吗？这个是有例子的[1]。 关于如何把kafka client的jar包配置到Python环境，分两种情况，当前有对应的两种解决方案： 1）如果是local运行，需要把kafka client的jar拷贝到python环境中pyflink的lib目录下 2）如果是remote运行，可以通过CLI的-j选项添加。 这两种方式对于Python用户来说可能都不太便捷，所以已有一个JIRA[3]在考虑添加另外一种对Python用户来说更友好的方式，欢迎到JIRA里参与讨论。 [1] https://ci.

Re: [ANNOUNCE] Apache Flink Python API(PyFlink) 1.9.2 released

Thanks for the great work, Jincheng. Regards, Dian > 在 2020年2月13日，下午1:32，jincheng sun 写道： > > Hi everyone, > > The Apache Flink community is very happy to announce the release of Apache > Flink Python API(PyFlink) 1.9.2, which is the first release to PyPI for the > Apache Flink Python API 1.

Re: [VOTE] Release Flink Python API(PyFlink) 1.9.2 to PyPI, release candidate #1

+1 (non-binding) - Verified the signature and checksum - Pip installed the package successfully: pip install apache-flink-1.9.2.tar.gz - Run word count example successfully. Regards, Dian > 在 2020年2月11日，上午11:44，jincheng sun 写道： > > > +1 (binding) > > - Install the PyFlink by `pip install` [

Re: [ANNOUNCE] Dian Fu becomes a Flink committer

Rui Li wrote: >> >>> Congratulations Dian, well deserved! >>> >>> On Thu, Jan 16, 2020 at 5:58 PM jincheng sun >>> wrote: >>> >>>> Hi everyone, >>>> >>>> I'm very happy to announce that Dian accepted

Re: 关于Window ProcessFunction数据丢失问题

window算子会丢弃迟到数据，可以看看是不是这个原因。如果是的话，可以调整一下watermark生成策略，适当增加点延迟，可以看一下：BoundedOutOfOrdernessTimestampExtractor > 在 2019年12月25日，上午10:39，1530130567 <1530130...@qq.com> 写道： > > 大佬们好： > > 最近使用window+processfunction处理数据，并加了watermark延迟处理。 > > 发现当数据input到达一定峰值后就会出现数据丢失的现象，请问是processfunction处理能力不够导致的吗？ >

Re: [DISCUSS] Drop Kafka 0.8/0.9

+1 for dropping them. Just FYI: there was a similar discussion few months ago [1]. [1] http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/DISCUSS-Drop-older-versions-of-Kafka-Connectors-0-9-0-10-for-Flink-1-10-td29916.html#a29997

Re: 流处理任务失败该如何追回之前的数据

如果使用的event time，watermark是根据event计算出来的，和系统时间没有关系，所以从最后一次checkpoint恢复即可。为什么你会觉得有问题？ > 在 2019年11月13日，下午8:29，柯桂强 写道： > > 我现在有一个流处理任务失败了，并且保留了checkpoint或者savepoint，我希望从最后一次checkpoint恢复，但是任务使用的是事件时间，超过窗口的数据就会被丢弃，我想到一个方法是，重启之前的数据通过批处理完成然后跑流处理，想问问大家这个方案是否可行，但是感觉如何限定批处理的范围并且和之后的流处理完美拼接是一个比较难的问题

Re: Flink1.7.2，TableApi的自定义聚合函数中如何使用自定义状态

佬，我所理解的对么？ > 请教大佬计算两个窗口之间的聚合值得差值这种场景在FlinkSql中实现的方案是啥？ > > From stevenchen > webchat 38798579 > > 发件人: Dian Fu<mailto:dian0511...@gmail.com> > 发送时间: Thursday, November 7, 2019 19:41 > 收件人: user-zh@flink.apache.org<mailto:user-zh@flink.apache.org> > 主题

Re: 广播状态是否可以设置ttl过期时间

是的。不过及时对于keyed state，如果你没有用TTL state这个功能，也是有可能返回过期的state的。 > 在 2019年11月8日，上午10:24，Yang Peng 写道： > > 嗯嗯，谢谢 付典老师，我理解的是虽然他不会删除但是也不应该返回过期的state值吧，应该是茶干老师说的那样只有keyed state 才支持 > state的ttl吧； > > Dian Fu 于2019年11月7日周四 下午8:08写道： > >> >> 1.8.0之前，ttl保证的是在ttl到达之前，数据不会清

Re: 广播状态是否可以设置ttl过期时间

1.8.0之前，ttl保证的是在ttl到达之前，数据不会清空，但是不保证ttl到达之后，数据一定清空。1.8.0之后提供了更完善的功能，可以看一下这个文章：https://flink.apache.org/2019/05/19/state-ttl.html > 在 2019年11月7日，下午3:06，Yang Peng 写道： > > > > -- Forwarded message - > 发件人： yangpengklf007

Re: Flink1.7.2，TableApi的自定义聚合函数中如何使用自定义状态

可以参考一下Flink代码里已有的例子：https://github.com/apache/flink/blob/c601cfd662c2839f8ebc81b80879ecce55a8cbaf/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/MaxWithRetractAggFunction.java

Re: flink1.9.1 on yarn sql 部署问题

问题1: 只需要在提交作业的机器上有flink即可 问题2: 两种方式都可以 问题3: 是的。lib目录下的jar和用户的jar会一起提交到YARN > 在 2019年10月30日，上午10:30，hb <343122...@163.com> 写道： > > hello: > > > 环境: flink1.9.1, on yarn hadoop2.6 > flink只安装在了一台提交的机器上, > > > lib目录下有文件: > flink-dist_2.11-1.9.1.jar > flink-json-1.9.0-sql-jar.jar > flink-shaded-ha

Re: Flink 的 log 文件夹下产生了 34G 日志

Henry 写道： > > > > 谢谢您。不过连接打不开呀。 > > http://apache-flink.147419.n8.nabble.com/Flink-BlobServerConnection-NoSuchFileException-td722.html#a724 > > > > > > > > 在 2019-10-28 20:02:21，"Dian Fu" 写道： >> 之前有过类似问题，你看一下这个回复，对你是否有帮助：

Re: Flink 的 log 文件夹下产生了 34G 日志

之前有过类似问题，你看一下这个回复，对你是否有帮助：http://apache-flink.147419.n8.nabble.com/Flink-BlobServerConnection-NoSuchFileException-td722.html#a724 > 在 2019年10月28日，下午3:54，Henry 写道： > > 您好！ > > > 请问一下，发现

  1   2   >