flink-user-zh mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jincheng sun <sunjincheng...@gmail.com>
Subject Re: 【PyFlink】对于数据以Csv()格式写入kafka报错,以及使用python udf时无法启动udf
Date Wed, 25 Mar 2020 03:16:28 GMT
Hi Zhefu,

谢谢您分享解决问题的细节,这对社区有很大的贡献!

1. 关于订阅问题

我想确认一下,你是否参考了[1],同时以订阅中文用户列表(user-zh@flink.apache.org)为例,您需要发送邮件到(
user-zh-subscribe@flink.apache.org),就是在原有邮件的地址上添加subscribe。同时收到一封“confirm
subscribe to *user-zh*@flink.apache.org”的确认邮件,需要进行确认回复。

2. 关于JAR包冲突问题

flink-python
JAR会携带flink-python对beam依赖的核心JAR包,我这里想了解一些,为啥你集群上面存在这beam相关的包?另外我认为您提供的case很好,让我想到了可以对PyFlink对Beam的依赖进行一些优化,比如将beam进行relocation.
我已经创建了社区改进JIRA[2].

3. 关于打包问题

上传给PyFlink的Python环境包需要是在机器间可移植的,所以的确不能包含软链接。如果是用virtualenv创建的环境的话,需要加上--always-copy选项。此外,如果集群机器上已经有准备好的python3.5+的环境,可以不用上传环境包,直接使用add_python_executable("python3")为集群指定要使用的Python
Interpreter。
除了virtualenv,conda/miniconda
也可用于创建虚拟环境,但是大小要大很多,在virtualenv处于某些原因不work的时候(比如源python解释器依赖的so文件在集群上不存在),可以考虑使用。

再次感谢您分享问题的解决细节和问题的反馈!

Best,
Jincheng

[1]
https://flink.apache.org/community.html#how-to-subscribe-to-a-mailing-list
[2] https://issues.apache.org/jira/browse/FLINK-16762


Zhefu PENG <pengzf0802@gmail.com> 于2020年3月24日周二 下午9:33写道:

> Hi Jincheng,
>
> 在中文邮件用户列表里我无法回复自己的问题(我已经十分确认subscribe了mailing
list, but i dont know why
> and how),所以在这里回复一下。 经过同事的帮忙和共同努力,我们初步解决了之前的疑问。反馈如下:
>
> 1. 首先是包冲突的问题,我们发现flink-python这个包下也有beam-runners-core-java-2.15.0.jar,
> beam-runners-direct-java-2.15.0.jar,  beam-runners-flink_2.11-2.15.0.jar,
> beam-sdk-java-core-2.15.0.jar这四个jar包的代码,若是运行的集群环境下本身也有这四个包的话,则会产生冲突,运行udf功能时找不到依赖。
>
> 2.
> 因为一些原因,集群的python默认环境无法改成python3,因此我们在代码中添加env.add_python_archive的功能,并且使用set_python_executor来帮助指定解释器。
> 用到的压缩包,在打包时一定要去掉软链接的使用,(可能在解压后不支持软链接查找。),我之前打包方式有错,所以出现了问题。
>
> 以上两点是我们排查出的原因,也终于能在cluster
> mode下成功运行udf。希望我们的反馈也能给你们的发展以及对用户的指导增加一点贡献。再次感谢帮助:)
>
> Best,
> Zhefu
>
> Zhefu PENG <pengzf0802@gmail.com> 于2020年3月20日周五 下午11:19写道:
>
>>
>> 感谢回复!beam就是直接用pip install方法安装的,因为在用pip install
>> apache-flink的时候发现有很多依赖,而且在安装时候要求安装beam的2.15.0的版本,我就安上了2.15.0。flink版本是1.10,没有从源码编译。因此我们也很困扰现在,希望能够求得帮助。
>>
>> 还有个小问题,就是我看到似乎你没有把日志一起转到邮件列表中,我想附带上去,那么我在订阅后,直接回复可以把日志贴上去吗?
>>
>> 顺祝,周末愉快:)非常感谢回复!
>>
>> 彭哲夫
>>
>> On Fri, Mar 20, 2020 at 22:34 jincheng sun <sunjincheng121@gmail.com>
>> wrote:
>>
>>> 彭哲夫,你好:
>>>
>>> 你是如何安装的beam,安装的版本是多少?如果是1.10 需要apache-beam
2.15,如果是 master需要apache-beam
>>> 2.19.
>>>
>>> BTW,  为了共享你的问题,我将你的问题发到了中文用户列表里面,我们大家一起讨论。
>>>
>>> Best,
>>> Jincheng
>>>
>>>
>>> Zhefu PENG <pengzf0802@gmail.com> 于2020年3月20日周五 下午5:12写道:
>>>
>>>> Hi Jincheng,
>>>>
>>>> 针对昨天提到的第二点,我做了两个思路:1. 将我的pyflink.py以本地模式运行,2.对集群节点进行环境配置
>>>>
>>>> 1.
>>>> 在本地模式中,我们发现在装apache-beam包的时候出于某些原因没有装全,少了_bz2模块,再补充上之后,pyflink.py的脚本可以正常运行,其中的udf功能也能正常使用。
>>>>
>>>> 2.
>>>> 在本地模式运行成功的基础上,我们根据你的建议,对所有的worker节点进行了环境的更新,都更新到了python3.6以及安装了apache_beam和apache-flink.
>>>> 但是以集群模式运行带有udf功能的脚本仍然报错,尝试谷歌搜索以后也没有搜到相关解答,在附件附上错误日志,希望能得到帮助(因为本地模式已经成功所以就不附带代码了),非常感谢!
>>>>
>>>> 期待您的回复
>>>> 彭哲夫
>>>>
>>>>
>>>> Zhefu PENG <pengzf0802@gmail.com> 于2020年3月19日周四 下午11:14写道:
>>>>
>>>>> Hi Jincheng:
>>>>>
>>>>> 非常感谢你如此迅速而细致的回复!~
>>>>>
>>>>> 关于第一点:根据你的回复,我在flink的lib目录下增加flink-csv-1.10.0-sql-jar.jar包之后,运行成功。而第一个包我在之前浏览你博客中关于kafka的使用的demo(based
>>>>> on flink 1.9)中有看到并下载,因此这里有个提议,或许你未来可以对于后续使用者补充
>>>>> flink-csv-1.10.0-sql-jar.jar包的使用的必要性 :),但也有可能是我在查询学习时看漏了,但不管怎么说感谢你的帮助解决;
>>>>>
>>>>> 关于第二点:因为部门相关组织安排问题,我现在没有权限去worker节点上查询,但是针对这一点我有个好奇的地方:我目前只在启动脚本的主机上安装了python3.5+,
>>>>> 并且除了udf功能外,我都能正常使用(比如sql本身就有的concat之类,或者add_columns()这种简单功能)。所以是不是我理解为,如果要使用pyflink的全部功能,应该是集群的环境都要是python3.5+?
>>>>> 但是简单的功能,只要启动脚本的主机环境符合就够了?
>>>>> 还是关于第二点,我刚刚又重新跑了一下脚本,本来是希望能获得和之前一样的错误日志发给我的mentor,但是发现这次报了新的问题:
>>>>> java.lang.NoClassDefFoundError: Could not initialize class
>>>>> org.apache.beam.sdk.options.PipelineOptionsFactory
>>>>>     at
>>>>> org.apache.flink.python.AbstractPythonFunctionRunner.open(AbstractPythonFunctionRunner.java:173)
>>>>>     at
>>>>> org.apache.flink.table.runtime.operators.python.AbstractPythonScalarFunctionOperator$ProjectUdfInputPythonScalarFunctionRunner.open(AbstractPythonScalarFunctionOperator.java:193)
>>>>>     at
>>>>> org.apache.flink.streaming.api.operators.python.AbstractPythonFunctionOperator.open(AbstractPythonFunctionOperator.java:139)
>>>>>     at
>>>>> org.apache.flink.table.runtime.operators.python.AbstractPythonScalarFunctionOperator.open(AbstractPythonScalarFunctionOperator.java:143)
>>>>>     at
>>>>> org.apache.flink.table.runtime.operators.python.PythonScalarFunctionOperator.open(PythonScalarFunctionOperator.java:73)
>>>>>     at
>>>>> org.apache.flink.streaming.runtime.tasks.StreamTask.initializeStateAndOpen(StreamTask.java:1007)
>>>>>     at
>>>>> org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$beforeInvoke$0(StreamTask.java:454)
>>>>>     at
>>>>> org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:94)
>>>>>
>>>>>
>>>>> 我猜测原因正如你提到的worker的环境不符标准,我会在明天上班后请同事帮忙check后,根据你的建议进行修改尝试。也希望能解答一下疑问,因为刚毕业参加工作,可能提的问题会显得比较低级,请见谅!
>>>>>
>>>>> 再次感谢你的回复,我会根据建议尽快进行错误修复
>>>>> 彭哲夫
>>>>>
>>>>> jincheng sun <sunjincheng121@gmail.com> 于2020年3月19日周四
下午9:08写道:
>>>>>
>>>>>> 彭哲夫,你好:
>>>>>>
>>>>>> 你上面问题可能原因是:
>>>>>>
>>>>>> 1. pyflink默认不包含kafka
>>>>>> connector的jar包和csv的格式JIR包,需要把这些jar包加到pyflink的lib目录下:
>>>>>>
>>>>>> $ PYFLINK_LIB=`python -c "import pyflink;import
>>>>>> os;print(os.path.dirname(os.path.abspath(pyflink.__file__))+'/lib')"`
>>>>>> $ cd $PYFLINK_LIB
>>>>>> $ curl -O
>>>>>> https://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-kafka_2.11/1.10.0/flink-sql-connector-kafka_2.11-1.10.0.jar
>>>>>> $ curl -O
>>>>>> https://repo1.maven.org/maven2/org/apache/flink/flink-csv/1.10.0/flink-csv-1.10.0-sql-jar.jar
>>>>>>
>>>>>> 2.
>>>>>> 有可能的原因是worker上没有安装python3以上环境或者环境中没有安装apache-beam,可以尝试在worker机器上执行一下:
>>>>>> python --version 检查python版本,同时执行 pip list 查看是否有apache-beam,如果没有,可以执行
>>>>>> :python -m pip install apache-flink
>>>>>>
>>>>>> 期望对你有帮助,有问题我们持续沟通。
>>>>>>
>>>>>> Best,
>>>>>> Jincheng
>>>>>>
>>>>>>
>>>>>>
>>>>>> Zhefu PENG <pengzf0802@gmail.com> 于2020年3月19日周四
下午8:13写道:
>>>>>>
>>>>>>> 你好:
>>>>>>>
>>>>>>>
>>>>>>> 在网上看到了你的博客,关于你对pyflink的开发和推动深感敬佩。我们部门因为业务需要最近在调研使用flink相关,我写了个一个简单的demo想做体验和测试,但是遇到了两个问题(第二个问题是目前遇到的比较大的困难,第一个问题采取了规避策略:)):
>>>>>>>
>>>>>>> 1.
>>>>>>> 当我把数据想以Csv格式输出到Kafka时,报错。(从社区文档我已经了解到应该用Csv()取代OldCsv(),并修改)。查看报错信息后我怀疑是因为缺少jar包导致(比如之前使用Json格式时候),但是从另一个文档中了解到csv格式应该是built-in的。目前采取了规避措施,采用json格式输出。
>>>>>>>
>>>>>>> 报错信息如下:
>>>>>>>
>>>>>>> py4j.protocol.Py4JJavaError: An error occurred while calling
>>>>>>> o62.insertInto.
>>>>>>> : org.apache.flink.table.api.NoMatchingTableFactoryException:
Could
>>>>>>> not find a suitable table factory for
>>>>>>> 'org.apache.flink.table.factories.SerializationSchemaFactory'
in
>>>>>>> the classpath.
>>>>>>>
>>>>>>> Reason: No factory supports all properties.
>>>>>>>
>>>>>>> The matching candidates:
>>>>>>> org.apache.flink.formats.csv.CsvRowFormatFactory
>>>>>>> Unsupported property keys:
>>>>>>> format.fields.#.name
>>>>>>> format.fields.#.data-type
>>>>>>>
>>>>>>> table注册部分代码如下:
>>>>>>> table_env.connect(Kafka()
>>>>>>>                   .version("universal")
>>>>>>>                   .topic(kafka_write_topic)
>>>>>>>                   .property(kafka_server,
>>>>>>> ','.join(kafka_server_list))
>>>>>>>                   .property(kafka_zookeeper,
>>>>>>> ','.join(kafka_server_list))) \
>>>>>>>     .with_format(Csv()
>>>>>>>                  .schema(DataTypes.ROW([DataTypes.FIELD("a",
>>>>>>> DataTypes.STRING()),
>>>>>>>                                         DataTypes.FIELD("b",
>>>>>>> DataTypes.STRING()),
>>>>>>>                                         DataTypes.FIELD("c",
>>>>>>> DataTypes.STRING())
>>>>>>>                                         ]))) \
>>>>>>>     .with_schema(Schema()
>>>>>>>                  .field("a", DataTypes.STRING())
>>>>>>>                  .field("b", DataTypes.STRING())
>>>>>>>                  .field("c", DataTypes.STRING())) \
>>>>>>>     .create_temporary_table(table_name_output)
>>>>>>>
>>>>>>> 2.采用规避策略后,尝试使用python udf增加自定义的函数丰富功能。但是在按照给的demo中的步骤定义好udf函数后,
>>>>>>> 才运行起来后,一段时间内会超时报错,猜测是因为pyflink-udf-runner.sh这个脚本没有被启用,
>>>>>>> 但是在所依赖的opt/flink-python_2.11-1.10.0.jar的包内可以找到该脚本。
>>>>>>> 报错信息如下:
>>>>>>> 2020-03-19 17:02:23,273 INFO
>>>>>>>  org.apache.beam.runners.fnexecution.environment.ProcessEnvironmentFactory
>>>>>>>  - Still waiting for startup of environment
>>>>>>> '/matrix/data/hadoop/data4/mapred/usercache/root/appcache/application_1584436506937_1231/python-dist-1e474e56-46a7-4ae6-bc4a-871ba86917a6/pyflink-udf-runner.sh'
>>>>>>> for worker id 1
>>>>>>>
>>>>>>> 这部分代码如下:
>>>>>>> @udf(input_types=[DataTypes.STRING()],
>>>>>>>      result_type=DataTypes.STRING())
>>>>>>> def if_length_enough(devid):
>>>>>>>     res = devid + " func"
>>>>>>>     return res
>>>>>>>
>>>>>>> table_env.register_function("if_length_enough", if_length_enough)
>>>>>>>
>>>>>>> table_env.from_path(table_name_input) \
>>>>>>>     .select("a, b, if_length_enough(c)") \
>>>>>>>     .insert_into(table_name_output)
>>>>>>>
>>>>>>>
>>>>>>> 以上两个错误,困扰了一下午,希望能帮忙解答,非常感谢!
>>>>>>> 期待您的回复。
>>>>>>>
>>>>>>> 彭哲夫
>>>>>>>
>>>>>>>
Mime
View raw message