Flink｜《Flink 官方文档 - 摆设 - 下令行界面 - 提交 PyFlink 作业》学习条记

学习文档：《Flink 官方文档 - 摆设 - 下令行界面 - 提交 PyFlink 作业》
学习条记如下：

---

当前，用户可以通过 CLI 提交 PyFlink 作业。对于通过 flink run 提交的 Python 作业，Flink 会实验 python 下令。因此，在启动 Python 作业前，必要先确定当前环境中的 python 下令指向 3.7+ 版本的 Python。
   示例：实验一个 PyFlink 作业

1. $ ./bin/flink run --python examples/python/table/word_count.py

复制代码

通过 --pyFiles 参数，可以添加 Python 的文件依靠，这个文件可以是 Python 文件、Python 包或其他文件。
   示例：实验一个 PyFlink 作业并添加依靠文件

1. $ ./bin/flink run \
2.    --python examples/python/table/word_count.py \
3.    --pyFiles file:///user.txt,hdfs:///$namenode_address/username.txt

复制代码

通过 --jarFile 参数，可以添加 Python 的 Jar 包依靠。在 --jarFile 参数中添加的 Jar 包将会上传到集群。
   实例：实验一个 PyFlink 作业并添加 Jar 包依靠

1. $ ./bin/flink run \
2.    --python examples/python/table/word_count.py \
3.    --jarfile <jarFile>

复制代码

通过 --pyModule 参数，可以以 module 模式实验 PyFlink 使命。
   示例：使用 module 模式实验 PyFlink 使命

1. $ ./bin/flink run \
2.    --pyModule word_count \
3.    --pyFiles examples/python/table

复制代码

通过 --jobmanager 参数，可以将 PyFlink 作业提交到指定的 JobManager。
   示例：将 PyFlink 使命提交到运行在 <jobmanagerHost> 的指定 JobManager

1. $ ./bin/flink run \
2.    --jobmanager <jobmanagerHost>:8081 \
3.    --python examples/python/table/word_count.py

复制代码

通过 --target 参数，可以使用 YARN Cluster in Per-Job Mode 实验 PyFlink 作业。
   示例：将 PyFlink 作业提交 YARN Cluster in Per-Job Mode

1. $ ./bin/flink run \
2.    --target yarn-per-job
3.    --python examples/python/table/word_count.py

复制代码

可以使用 run-application -t yarn-application 下令将 PyFlink 作业提交到 YARN cluster in Application Mode。此中，通过 -pyarch 指定的存档文件将通过 blob 服务器分发到 TaskManagers，此中文件巨细限定为 2GB。假如存档文件的巨细高出 2GB，则可以先将其上传到分布式文件体系，然后使用下令行选项 -pyarch 指定路径。
   示例：将 PyFlink 作业提交到 YARN cluster in Application Mode

1. $ ./bin/flink run-application -t yarn-application \
2.    -Djobmanager.memory.process.size=1024m \
3.    -Dtaskmanager.memory.process.size=1024m \
4.    -Dyarn.application.name=<ApplicationName> \
5.    -Dyarn.ship-files=/path/to/shipfiles \
6.    -pyarch shipfiles/venv.zip \
7.    -pyclientexec venv.zip/venv/bin/python3 \
8.    -pyexec venv.zip/venv/bin/python3 \
9.    -pyfs shipfiles \
10.    -pym word_count

复制代码

在这个使命中，假定 Python 依靠位于 /path/to/shipfiles 中；比方，此中应该包罗 venv.zip 和 word_count.py。
  当它在 YARN Application Mode 下的 JobManager 实验作业时，-pyarch 和 pyfs 中指定的路径是相对于 shipfiles 的路径，shipfiles 是已发送文件的目次名称。
  可以使用 run-application -target kubernetes-application 下令将 PyFlink 作业提交到 native Kubernetes cluster，这必要一个已经安装了 PyFlink 的 Docker 镜像。
   示例：将 PyFlink 作业提交到 native Kubernetes cluster

1. $ ./bin/flink run-application \
2.    --target kubernetes-application \
3.    --parallelism 8 \
4.    -Dkubernetes.cluster-id=<ClusterId> \
5.    -Dtaskmanager.memory.process.size=4096m \
6.    -Dkubernetes.taskmanager.cpu=2 \
7.    -Dtaskmanager.numberOfTaskSlots=4 \
8.    -Dkubernetes.container.image.ref=<PyFlinkImageName> \
9.    --pyModule word_count \
10.    --pyFiles /opt/flink/examples/python/table/word_count.py

复制代码

在 run 下令行 run-application 下令中支持的参数清单：

• -py / --python：Python 步调入口脚本，其依靠可以通过 --pyFiles 参数添加。
  
• -pym / --pyModule：Module 模式的 Python 步调入口，这个参数必须与 --pyFiles 同时使用。
  
• -pyfs / --pyFiles：为作业附加自界说文件，这些文件将被添加到当地客户端以及远端 Python UDF 的 PYTHONPATH 中；此中后缀为 .zip 的文件会被提取出来并添加到 PYTHONPATH 中；多个文件之间可以试用 , 分隔。比方 --pyFiles file:///tmp/myresource.zip,hdfs:///$namenode_address/myresource2.zip。
  
• -pyarch / --pyArchives：为作业添加 Python 存档文件；这些存档文件会被提取到 Python UDF Worker 的工作目次中。比方 --pyArchives file:///tmp/py37.zip,file:///tmp/data.zip#data --pyExecutable py37.zip/py37/bin/python。
  
• -pyclientexec / --pyClientExecutable：在提交 Python 使命时，以及编译 Python UDF 时使用的 Python 表明器路径。比方 --pyArchives file:///tmp/py37.zip --pyClientExecutable py37.zip/py37/python。
  
• -pyexec / --pyExecutable：实验 Python UDF 的 Python 表明器路径。比方 --pyExecutable /usr/local/bin/python3。
  
• -pyreq / --pyRequirements：指定作业所需第三方模块的 requirements.txt 文件；这些文件将被安装并添加到 Python UDF 的 PYTHONPATH 中。可以选择指定一个目次，用于存储这些依靠的安装包，假如指定了目次，则使用 # 作为分隔符。比方 --pyRequirements file:///tmp/requirements.txt#file:///tmp/cached_dir.