Pyflink如何使用docker部署的flink服务

我通过docker部署了flink，想使用python-API进行操作提交py任务。在flink容器内是没有python环境的，我该如何进行操作才能使两者能够互联互通。同时如果我在其他服务器上部署的python环境，编写相关的job（如wordcount.py）我该通过何种方式提交？

该回答引用ChatGPT

在使用 PyFlink 前，需要确保 Flink 集群已经在 Docker 中成功启动，并且正在运行。在 Flink Web UI 中可以查看 Flink 集群状态，确保它已经启动并处于运行状态。

1、接下来，需要将 PyFlink 添加到 Python 环境中。可以使用 pip 安装 PyFlink，命令如下：

pip install apache-flink

2、安装完成后，就可以使用 PyFlink 编写 Flink 任务。首先需要创建一个 PyFlink 环境，代码如下：

from pyflink.dataset import ExecutionEnvironment

env = ExecutionEnvironment.get_execution_environment()

3、然后可以编写 Flink 任务，比如一个简单的 Word Count 程序，代码如下：

from pyflink.dataset import ExecutionEnvironment

env = ExecutionEnvironment.get_execution_environment()

data = env.from_elements("Hello World", "Hello PyFlink", "Hello PyFlink")

result = data \
    .flat_map(lambda s, collector: [word for word in s.lower().split()]) \
    .map(lambda word, count: (word, 1)) \
    .group_by(0) \
    .sum(1)

result.print()

4、最后需要将任务提交到 Flink 集群上运行，代码如下：

from pyflink.datastream import StreamExecutionEnvironment

env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)

data = env.from_collection([(1, "Hello"), (2, "World")])

data.map(lambda i, s: (i, s.upper())).print()

env.execute("pyflink demo")

5、在上述代码中，需要设置 Flink 集群的 IP 地址和端口，这可以通过设置 PyFlink 环境变量 FLINK_REST_URL 来实现。代码如下：

import os

os.environ["FLINK_REST_URL"] = "http://flink_jobmanager:8081"

其中，"flink_jobmanager" 是 Docker 中 Flink 集群的容器名，可以通过 docker ps 命令查看。端口号是 Flink 集群 Web UI 的端口号，默认为 8081。

6、最后需要运行 Python 程序，命令如下：

python pyflink_demo.py

这将会把 PyFlink 任务提交到 Docker 中的 Flink 集群上运行。可以在 Flink Web UI 中查看任务状态和结果。

该回答引用ChatGPT
首先，如果您要提交Python任务，首先您需要在Flink容器中安装Python环境，比如Anaconda。通过如下命令可以安装：

sudo apt-get install curl
curl -O https://repo.anaconda.com/archive/Anaconda3-2020.07-Linux-x86_64.sh
bash Anaconda3-2020.07-Linux-x86_64.sh

安装完成后，即可使用Python对Flink进行操作。另外，您还可以通过如下命令来提交Python Job：

./bin/flink run --py pyJob.py

查看Flink 官网PyFlink的环境安装：