目錄

背景

一、pyflink安裝

二、編寫demo程序

三、提交yarn前準備

?四、提交任務

五、踩坑記錄

1、提交任務時客戶端出現語法錯誤

2、提交任務時客戶端出現lzma包找不到

3、提交任務時客戶端出現“org.apache.flink.streaming.api.utils.PythonTypeUtils.getCollectionInputFormat does not exist in the JVM”

4、提交任務時taskmanager上出現找不到pyflink

5、提交任務時taskmanager上報錯出現Permission denied

六、總結

---

背景

? ? ? ? 最近項目需要，學習研究使用python來開發flink任務，在此做相關筆記。

一、pyflink安裝

? ? ? ? 在本地執行如下命令，需要注意的是，pyflink必須要求python版本大于等于3.6，我本地是3.12

創建虛擬環境

python -m venv venv

切換虛擬環境

venv/bin/activate

執行安裝命令

pip install apache-flink==1.14

?安裝成功后如下

二、編寫demo程序

? ? ? ? 這里直copy官方教程

import argparse
import logging
import sysfrom pyflink.common import Row
from pyflink.table import (EnvironmentSettings, TableEnvironment, TableDescriptor, Schema,DataTypes, FormatDescriptor)
from pyflink.table.expressions import lit, col
from pyflink.table.udf import udtfword_count_data = ["To be, or not to be,--that is the question:--","Whether 'tis nobler in the mind to suffer","The slings and arrows of outrageous fortune","Or to take arms against a sea of troubles,","And by opposing end them?--To die,--to sleep,--","No more; and by a sleep to say we end","The heartache, and the thousand natural shocks","That flesh is heir to,--'tis a consummation","Devoutly to be wish'd. To die,--to sleep;--","To sleep! perchance to dream:--ay, there's the rub;","For in that sleep of death what dreams may come,","When we have shuffled off this mortal coil,","Must give us pause: there's the respect","That makes calamity of so long life;","For who would bear the whips and scorns of time,","The oppressor's wrong, the proud man's contumely,","The pangs of despis'd love, the law's delay,","The insolence of office, and the spurns","That patient merit of the unworthy takes,","When he himself might his quietus make","With a bare bodkin? who would these fardels bear,","To grunt and sweat under a weary life,","But that the dread of something after death,--","The undiscover'd country, from whose bourn","No traveller returns,--puzzles the will,","And makes us rather bear those ills we have","Than fly to others that we know not of?","Thus conscience does make cowards of us all;","And thus the native hue of resolution","Is sicklied o'er with the pale cast of thought;","And enterprises of great pith and moment,","With this regard, their currents turn awry,","And lose the name of action.--Soft you now!","The fair Ophelia!--Nymph, in thy orisons","Be all my sins remember'd."]def word_count(input_path, output_path):t_env = TableEnvironment.create(EnvironmentSettings.in_streaming_mode())# write all the data to one filet_env.get_config().get_configuration().set_string("parallelism.default", "1")# define the sourceif input_path is not None:t_env.create_temporary_table('source',TableDescriptor.for_connector('filesystem').schema(Schema.new_builder().column('word', DataTypes.STRING()).build()).option('path', input_path).format('csv').build())tab = t_env.from_path('source')else:print("Executing word_count example with default input data set.")print("Use --input to specify file input.")tab = t_env.from_elements(map(lambda i: (i,), word_count_data),DataTypes.ROW([DataTypes.FIELD('line', DataTypes.STRING())]))# define the sinkif output_path is not None:t_env.create_temporary_table('sink',TableDescriptor.for_connector('filesystem').schema(Schema.new_builder().column('word', DataTypes.STRING()).column('count', DataTypes.BIGINT()).build()).option('path', output_path).format(FormatDescriptor.for_format('canal-json').build()).build())else:print("Printing result to stdout. Use --output to specify output path.")t_env.create_temporary_table('sink',TableDescriptor.for_connector('print').schema(Schema.new_builder().column('word', DataTypes.STRING()).column('count', DataTypes.BIGINT()).build()).build())@udtf(result_types=[DataTypes.STRING()])def split(line: Row):for s in line[0].split():yield Row(s)# compute word counttab.flat_map(split).alias('word') \.group_by(col('word')) \.select(col('word'), lit(1).count) \.execute_insert('sink') \.wait()# remove .wait if submitting to a remote cluster, refer to# https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/python/faq/#wait-for-jobs-to-finish-when-executing-jobs-in-mini-cluster# for more detailsif __name__ == '__main__':logging.basicConfig(stream=sys.stdout, level=logging.INFO, format="%(message)s")parser = argparse.ArgumentParser()parser.add_argument('--input',dest='input',required=False,help='Input file to process.')parser.add_argument('--output',dest='output',required=False,help='Output file to write results to.')argv = sys.argv[1:]known_args, _ = parser.parse_known_args(argv)word_count(known_args.input, known_args.output)

在本地執行該腳本，輸出如下

三、提交yarn前準備

準備一臺linux服務器，并裝有flink客戶端（我使用的版本是flink1.14.2，這里不說客戶端如何安裝了，下載包解壓安裝即可）

在服務器上搭建pyflink運行環境，參考第一章節

將demo程序上傳到該服務器上

其中env為python虛擬環境目錄，py_env.zip為將env使用zip進行壓縮的文件

?四、提交任務

/home/master/flink-1.14.2/bin/flink run -pyarch py_env.zip -m yarn-cluster -py /home/zhubao/demo.py -pyexec py_env.zip/env/bin/python

看到終端打印如下日志

訪問yarn集群web管理頁面，在running下看到有對應的任務時，即表示任務已經提交到yarn集群

查看任務詳情

五、踩坑記錄

1、提交任務時客戶端出現語法錯誤

SLF4J: Actual binding is of type [org.apache.logging.slf4j.Log4jLoggerFactory]File "main.py", line 55ds.print()^
SyntaxError: invalid syntax

解決方法：上述問題排查發現是flink客戶端版本差異導致，編寫的demo代碼，和flink客戶端版本要一致，否則會出現一些莫名其妙的問題，統一調整flink版本為一致，包括pyflink，flink客戶端等

2、提交任務時客戶端出現lzma包找不到

Traceback (most recent call last):File "/home/zhubao/env/lib/python3.7/site-packages/fastavro/read.py", line 2, in <module>from . import _readFile "fastavro/_read.pyx", line 11, in init fastavro._readFile "/home/master/python3/lib/python3.7/lzma.py", line 27, in <module>from _lzma import *
ModuleNotFoundError: No module named '_lzma'

解決方法：該錯誤表明系統缺少Python的LZMA壓縮模塊依賴（_lzma），這是Python標準庫中處理.xz壓縮文件的底層C擴展模塊，需要進行安裝

# 使用root權限執行如下命令
yum install -y xz-devel python-backports-lzma

# 使用普通用戶執行安裝lzma包命令
pip install backports.lzma -i https://pypi.tuna.tsinghua.edu.cn/simple --trusted-host pypi.tuna.tsinghua.edu.cn

安裝完成后，需要對lzma文件進行修改，找到lzma.py文件，一般在$PYTHON_HOME/lib/python3.7目錄下（根據實際版本），主要是加上try except

修改完成后保存退出，重新執行解決該問題

3、提交任務時客戶端出現“org.apache.flink.streaming.api.utils.PythonTypeUtils.getCollectionInputFormat does not exist in the JVM”

Traceback (most recent call last):File "main.py", line 99, in <module>word_count(known_args.input, known_args.output)File "main.py", line 39, in word_countds = env.from_collection(word_count_data)File "/home/zhubao/env/lib/python3.7/site-packages/pyflink/datastream/stream_execution_environment.py", line 958, in from_collectionreturn self._from_collection(collection, type_info)File "/home/zhubao/env/lib/python3.7/site-packages/pyflink/datastream/stream_execution_environment.py", line 981, in _from_collectionj_input_format = PythonTypeUtils.getCollectionInputFormat(File "/home/zhubao/env/lib/python3.7/site-packages/py4j/java_gateway.py", line 1550, in __getattr__"{0}.{1} does not exist in the JVM".format(self._fqn, name))
py4j.protocol.Py4JError: org.apache.flink.streaming.api.utils.PythonTypeUtils.getCollectionInputFormat does not exist in the JVM

解決方法：還是版本不匹配導致，請確保pyflink與客戶端版本一致

4、提交任務時taskmanager上出現找不到pyflink

Caused by: java.io.IOException: Failed to execute the command: python -c import pyflink;import os;print(os.path.join(os.path.abspath(os.path.dirname(pyflink.__file__)), 'bin'))
output: Traceback (most recent call last):File "<string>", line 1, in <module>
ImportError: No module named pyflink

解決方法：在flink配置文件中，加上python執行環境，打開flink-conf.yaml文件，一般在$FLINK_HOME/conf目錄下，編輯該文件，在末尾加上python執行環境

python.client.executable: /home/zhubao/env/bin/python

5、提交任務時taskmanager上報錯出現Permission denied

Caused by: java.io.IOException: Cannot run program "/home/zhubao/env/bin/python": error=13, Permission denied

解決方法：找了一些方案，但最終通過將整個python執行環境打包提交到yarn上解決，方法如下

將python環境達成zip包

zip -r py_env.zip /home/zhubao/env/

提交任務命令增加指定環境包與執行環境

/home/master/flink-1.14.2/bin/flink run -pyarch py_env.zip -m yarn-cluster -py /home/zhubao/demo.py -pyexec py_env.zip/env/bin/python

六、總結

? ? ? ? 以上是使用pyflink進行flink任務開發，以及將任務提交到yarn集群方法。