前言

为了能更方便的开发flink应用, 笔者决定使用python+flink进行代码开发,然而中文论坛上很难找到一个详细的教程在ubuntu上配置一个pyflink集群,用于提交python脚本执行任务。笔者在提交pyflink任务的时候踩了很多坑，最后终于配好了框架并且能够提交python任务，归根结底造成这些困难的原因还是环境没搭好,希望这篇文章能够给初学者带来一些帮助,节省时间快速入门。
首先,有问题一定要首先从官方获取帮助pyflink官方文档 。
其次，要学会看错误输出，首先自己粗略浏览一遍错误输出，看到有没有熟悉的内容是自己有操作过，配置过的，比如一些IP地址，还有python解释器的版本，这些都是可以自行更改配置的，通常如果一个刚搭建的flink框架如果有问题，那么应该着重排查这些可以自行配置的内容。

开始之前

环境准备

这里的环境列出的是笔者所用的环境，大家可以根据自己的需求使用，其中python的版本应在3.7~3.10之间，后续如果支持到更高的python版本，请以官方文档为准。

环境	版本
jdk	1.8.0_162
python	3.8.10
apache-flink	1.17.1
Ubuntu	ubuntu1~20.04

apache-flink就是我们要安装的pyflink，它不仅包含了python的API，也包含了一个完整的flink框架，我们需要使用这个框架来启动集群。
有的同学可能希望通过anaconda来管理环境，值得一提的是，pyflink并不能通过conda install进行安装，笔者由于并不熟练使用conda管理虚拟环境，不知道如何解决后面提到的链接操作，导致踩了坑，所以笔者也不推荐初学者上来看到flink官方说可以使用anaconda管理虚拟环境，于是也配了anaconda。如果使用了anaconda创建了虚拟环境，并且能够让flink集群正确用上合适的python环境，那么可以通过调用对应python环境下的pip来安装pyflink。

安装JDK

由于笔者使用的是oracle的jdk,所以笔者建议使用oracle的jdk而不是openjdk。
如果已经安装了jdk，这部分内容可以跳过。
首先把准备好的jdk文件解压到你想要的目录，这里我使用的是

/usr/lib/jvm/

解压文件:

sudo tar -zxvf ./jdk-8u162-linux-x64.tar.gz -C /usr/lib/jvm

配置JAVA_HOME

vim ~/.bashrc

添加以下内容:

export JAVA_HOME=/usr/lib/jvm/jdk1.8.0_162
export JRE_HOME=${JAVA_HOME}/jre
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib
export PATH=${JAVA_HOME}/bin:$PATH

应用更改

source ~/.bashrc

测试java环境:

java -version

如果出现:

则说明安装成功,继续下一步

安装python3.8

如果已经安装过python3.7~3.10,那么可以跳过安装本体部分
如果不确定自己是否已经安装过python,可以使用以下命令,查看目前可用的python:

ls /usr/bin/python*

这条命令查看了系统能够调用的python二进制文件的软链接。
值得注意的是,pyflink在提交任务的时候,使用的是/usr/bin/文件夹下名为python的软连接,所以这个名为python的软连接一定要配置,并且包含了apache-flink所需要的库。
简单来说,如果你在命令行中输入python能够打开python的命令行,并且可以正常调用apache-flink库,那么flink就可以调用python解释器接收python脚本形式的任务。

安装python本体

(选择性操作)删除原来的python链接

如果选择重新安装python,安装了新的环境,想要使用新的环境,则需要删除原来的链接,并添加新链接

sudo rm /usr/bin/python3   

正式安装python3

文章参考:Ubuntu安装Python3.8

sudo apt-get update
sudo apt-get install software-properties-common
sudo add-apt-repository ppa:deadsnakes/ppa
sudo apt-get update
sudo apt-get install python3.8

为python添加链接

sudo ln -s /usr/bin/python3.8 /usr/bin/python3

这时候可以尝试使用来验证python是否正确配置

python3 --version

然而链接的添加并没有结束,根据pyflink官方文档的提示，

我们需要添加：

sudo ln -s /usr/bin/python3 /usr/bin/python

使用pip安装所需的环境

(选择性操作)安装pip

默认安装Python3之后自动安装pip3，如果在终端中敲入

pip3 -V

显示

bash: pip command not found

那么需要设置一下pip

sudo apt-get install python3-pip

安装pyflink

如果pip已经可以正常使用,那么直接输入:

pip install apache-flink

笔者推荐使用pip官方源,如果有下载速度过慢,连接超时的问题,可以使用代理解决。
其中我们只需要在主机端的clash找到主机的IP地址，和clash配置的端口号，组成这样的形式:http://主机IP:端口号 的形式，就可以使用主机上的代理访问官方源。
使用代理的pip 指令：

pip install apache-flink --proxy=http://主机IP:端口号

安装过程会自动安装依赖包，如果安装过程中有红字提示依赖未安装，可以手动安装。

使用pyflink

如果上一步没有安装出错，那么我们可以使用

pip show apache-flink

来查看pyflink的安装位置：

进入flink的文件夹：

cd /home/hadoop/.local/lib/python3.8/site-packages/pyflink/

我们可以看到文件夹下有完整的flink框架：

通过examples测试一下flink的pythonAPI:

python examples/datastream/word_count.py

看到如下输出，并且没有报错。
说明flink pythonAPI调用成功。

单节点运行以及任务提交测试

配置flink-conf.yaml

在启动节点之前，首先要配置./conf/flink-conf.yaml的参数

 vim ./conf/flink-conf.yaml

其中

jobmanager.rpc.address: Master

Master是笔者的taskmanager节点名称，也是本机名称，大家可以根据自己的实际情况进行修改
其余参数
如果需要使用flinkweb，可以把rest.bind-address简单配置成：

rest.bind-address: 0.0.0.0

启动节点

启动集群，以提交任务

./bin/start-cluster.sh

可以使用jps查看节点是否成功启动:

提交任务

提交的任务我们采用examples，这样就不需要自己编写代码，也不需要准备数据源

./bin/flink run --python ./examples/datastream/word_count.py

可以看到虽然没有输出，但是也没有报错，没有输出，就是最好的输出。
到此我们已经完成了flink单节点任务提交测试，后续会更新集群的配置以及测试

集群配置以及测试

配置hostnames

对网络比较熟悉的同学可以跳过这一部分。
为了能够使用别名来连接各个节点，方便配置文件的编写，笔者建议修改hosts文件，用别名来代替IP地址
首先查看当前节点的IP地址：

ip a

一般来说只会有一个网卡，笔者这里配置了两个网卡，挑选一个能够让节点之间互相ping通的ip地址即可

修改hosts配置文件：

sudo vim /etc/hosts

添加以下内容（根据自己的实际情况修改）：

192.168.56.2 Master
192.168.56.3 Slave1
192.168.56.4 Slave2

测试：

这里演示了Master的配置方法，其余的节点同样地添加以上的三行内容即可。

设置主节点FLINK_HOME

FLINK_HOME是一个环境变量参数，我们可以在用户目录（建议）下的.bashrc 文件配置它。
这里我们首先配置MASTER的FLINK_HOME。
FLINK_HOME表示了当前机器的Flink位置，
同时FLINK_HOME告诉MASTER（主节点）如何从子节点找到Flink可执行程序并启动它，成为worker。
所以Worker的Flink路径应该要和Master的Flink路径一致。

查找pyflink的安装位置以及相关参数

pip show apache-flink

这里我们可以看到一会要用到的flink的版本号以及安装位置等信息。

version:1.17.1
location:/home/hadoop/.local/lib/python3.8/site-packages

修改.bashrc

我们可以先试着切换到flink的文件夹，确认一下文件是否存在：

cd  /home/hadoop/.local/lib/python3.8/site-packages/pyflink/

如果存在，复制这个路径（ /home/hadoop/.local/lib/python3.8/site-packages/pyflink/），一会要用到

vim ~/.bashrc

添加：

export FLINK_HOME=/home/hadoop/.local/lib/python3.8/site-packages/pyflink
export PATH=$FLINK_HOME/bin:$PATH

笔者由于worker的环境是python3.10，且没办法卸载后安装3.8，经过在worker测试集群后，最后定下了这个目录，
但是理论上用3.8的路径也是可以的。

保存之后，输入：

source ~/.bashrc 
flink --version

如果配置正确，那么终端会显示flink的版本，否则会报错

配置子节点

在分发flink之前，需要按照上面的内容对其它节点配置，但是中间如果有python环境不一样，比如主节点用的是python3.8而子节点用的是python3.10，这种情况不需要惊慌，虽然有可能因为python的API不一致导致flink启动失败，但其实保证flink版本一致，其实问题不大。
值得注意的是，即便我们要从Master拿一个flink客户端来用，也依旧需要用pip安装apache-flink 来安装相关的依赖。
子节点也需要配置FLINK_HOME，

检查flink相关依赖以及版本

在配置之前，我们应该检查pip安装的flink和主机的flink版本是否一致。

pip show apache-flink

也可以切换到pyflink目录下,输入：

./bin/flink --version

查看版本

如果版本不一致，则使用

pip uninstall apache-flink
pip install apache-flink==1.17.1

安装指定版本的flink。

还有一个问题是有的同学可能worker用的python版本和Master的不一样，比如笔者Master用的是3.8，而worker用的是3.10，这时候我们只要

mkdir -p /home/hadoop/.local/lib/python3.8/site-packages/pyflink/

来创建这个目录就好。

修改Master的flink配置文件

flink-conf.yaml

flink-conf.yaml是flink客户端的主要配置文件，大部分运行参数都在这里配置。
进入master的FLINK目录

cd $FLINK_HOME

编辑flink.conf：

vim ./conf/flink-conf.yaml

根据自己的实际情况，修改以下参数：

jobmanager.rpc.address: Master
io.tmp.dirs: /home/hadoop/.local/lib/python3.8/site-packages/pyflink/temp
jobmanager.bind-host: 0.0.0.0
taskmanager.bind-host: 0.0.0.0
rest.bind-address: 0.0.0.0
taskmanager.host: Master

其中
jobmanager.rpc.address：jobmanager的IP地址。
io.tmp.dirs:临时文件的存放位置。
然后为io.tmp.dirs创建相应的路径，以免后面因为找不到路径抛出异常。

mkdir /home/hadoop/.local/lib/python3.8/site-packages/pyflink/temp

masters

./conf/masters文件声明了集群的master，也就是jobmanager。

vim ./conf/master

添加以下内容：

master:8081

workers

./conf/wokers文件声明了集群中的taskermanagers。

vim ./conf/workers

添加以下内容：

Master
Slave1
Slave2

打包flink并分发

为了分发flink，我们需要先对其打包

cd ..
tar -zcf  ~/pyflink.master.tar.gz ./pyflink

这条命令把flink打包为pyflink.master.tar.gz ，并存放于当前用户的home文件夹下。

分发flink

scp ./pyflink.master.tar.gz Slave1:/home/hadoop
scp ./pyflink.master.tar.gz Slave2:/home/hadoop

接下来我们到其他节点进行操作：
Slave1：

mkdir -p /home/hadoop/.local/lib/python3.8/site-packages/
sudo tar -zxf ~/pyflink.master.tar.gz -C /home/hadoop/.local/lib/python3.8/site-packages/
chown -R hadoop /home/hadoop/.local/lib/python3.8/site-packages/pyflink/

Slave2：

mkdir -p /home/hadoop/.local/lib/python3.8/site-packages/
sudo tar -zxf ~/pyflink.master.tar.gz -C /home/hadoop/.local/lib/python3.8/site-packages/
chown -R hadoop /home/hadoop/.local/lib/python3.8/site-packages/pyflink/

这里包含了解压（ tar -zxf ）和修改路径下所有文件所属权（chown -R ）
修改文件所属权很重要，因为文件来源是另外一个节点，所以解压出来的文件权限不属于当前节点的用户，会导致无法写入。

启动集群

在Master上操作：

cd /home/hadoop/.local/lib/python3.8/site-packages/pyflink/
./bin/start-cluster.sh

如果没有报错，我们可以等待一段时间（大概3~5分钟）
然后在每个节点上查看JPS：

jps

master:

slave1:

slave2:

如果看到对应的进程都已经正常启动，且保持到了3~5分钟左右，就可以确定现在集群已经正常运行了
接下来再检查一下slot数目是否正常：
在浏览器输入：

http://192.168.56.2:8081/

这里的192.168.56.2是master的IP地址，大家根据自己的实际情况填写
按照配置文件，我的每个节点提供一个slot，所以总共有3个slot：

如果大家的slot和自己的节点数不一致，那么请检查一下子节点flink的配置文件是否和主机的一致。

另外，最后笔者还想确认一点：
master在子节点中启动的flink真的是是根据FLINK_HOME启动的吗？
笔者移动了各节点的的pyflink文件夹到python3.10（原本在3.8下）的目录下，并且修改了相应的FLINK_HOME，
在启动flink集群之后，我们在master的命令行输入:

jps -lv

这条命令用于查看jvm下运行的程序的类的路径等。
可以看到：
master：

slave1：

slave2：

这个结果说明，FLINK_HOME告诉了master，flink在子节点的哪个位置。

笔者遇到过的坑

坑1：代理

之前为了apt能够使用代理，在Ubuntu桌面的设置里配置了代理，导致报错，仔细检查了日志，发现有里面有配置的代理的IP地址，虽然当时不知道是不是这个问题导致无法提交任务上flink，但是确实给我造成了烟雾弹的效果。
关闭ubuntu代理（桌面）：

关闭ubuntu代理（terminal）：
之前还在bashrc里配置了代理，也先注释掉了

vim ~/.bashrc 

设置完后重启
还有其他情况，比如直接在命令行输入export http_proxy 来设置代理的，直接重启机器就好。

坑2：没有配置python链接

在笔者踩其他坑之前，这个坑可能一直影响着笔者，导致笔者一直不能成功上传任务，如果笔者早些按照官方文档的说明来配置flink，也许不需要踩那些坑。

坑3：运行着Scala的flink，上传py的任务

当时并不知道安装pyflink会自带一个flink框架，用了以前学习的时候用的老flink，所以倒腾了半天也没有进度。

坑4：权限问题

分发flink的时候，没有更改路径的权限，启动不起来

结语

待施工