【大数据】Flink 命令行详细操作指南

Flink 的任务在 hdfs 上生成的资源

[root@spark-31 hadoop-3.3.1]# bin/hdfs dfs -ls /user/root/.flink/application_1667200821707_0070/
Found 7 items
-rw-r--r--   2 root supergroup      16878 2022-12-20 16:01 /user/root/.flink/application_1667200821707_0070/TopSpeedWindowing.jar
-rw-r--r--   2 root supergroup       1039 2022-12-20 16:01 /user/root/.flink/application_1667200821707_0070/application_1667200821707_0070-flink-conf.yaml5686370271306099143.tmp
-rw-r--r--   2 root supergroup      43807 2022-12-20 16:01 /user/root/.flink/application_1667200821707_0070/application_1667200821707_00705692020442525497249.tmp
-rw-r--r--   2 root supergroup  117099297 2022-12-20 16:01 /user/root/.flink/application_1667200821707_0070/flink-dist-1.16.0.jar
drwx------   - root supergroup          0 2022-12-20 15:47 /user/root/.flink/application_1667200821707_0070/lib
-rw-r--r--   2 root supergroup       2694 2022-12-20 15:47 /user/root/.flink/application_1667200821707_0070/log4j.properties
drwx------   - root supergroup          0 2022-12-20 16:00 /user/root/.flink/application_1667200821707_0070/plugins

作业提交

Flink 与多个集群管理框架（如 Kubernetes 或 YARN）兼容，这些框架在资源提供程序部分有更详细的描述。可以在不同的部署模式下提交作业。作业提交的参数化因基础框架和部署模式而异。
bin/flink 提供了一个参数 --target 来处理不同的选项。除此之外，还必须使用run（对于Session and Per-Job Mode）或run-application（对于 Application Mode）提交作业。请参阅以下参数组合摘要：

• YARN
  ./bin/flink run --target yarn-session: 提交到已在 YARN 集群上运行的 Flink
  ./bin/flink run --target yarn-per-job: 提交在per-job模式下在 YARN 群集上启动 Flink
  ./bin/flink run-application --target yarn-application: 提交在应用程序模式下在 YARN 集群上启动 Flink
• Kubernetes
  ./bin/flink run --target kubernetes-session: 提交到 Kubernetes 集群上已经运行的 Flink
  ./bin/flink run-application --target kubernetes-application: 提交在应用程序中的 Kubernetes 集群上启动 Flink
• Standalone:
  ./bin/flink run --target local:在会话模式下使用微型集群进行本地提交
  ./bin/flink run --target remote: 提交到已在运行的 Flink 集群

提交 PyFlink 作业

以下命令显示了不同的 PyFlink 作业提交用例

• Run a PyFlink job:

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

使用其他源文件和资源文件运行 PyFlink 作业。在 --pyFiles 中指定的文件将被添加到 PYTHONPATH 中，因此在 Python 代码中可用。

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

运行将引用 Java UDF 或外部连接器的 PyFlink 作业。–jarfile 中指定的 JAR 文件将上传到集群。

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

使用 pyFiles 和 --pyModule 中指定的主入口模块运行 PyFlink 作业：

./bin/flink run \
      --pyModule table.word_count \
      --pyFiles examples/python/table

在主机上运行的特定 JobManager 上提交 PyFlink 作业（相应地调整命令）：

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

• Run a PyFlink job using a YARN cluster in Per-Job Mode:

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

• Run a PyFlink job using a YARN cluster in Application Mode:

./bin/flink run-application -t yarn-application \
      -Djobmanager.memory.process.size=1024m \
      -Dtaskmanager.memory.process.size=1024m \
      -Dyarn.application.name=<ApplicationName> \
      -Dyarn.ship-files=/path/to/shipfiles \
      -pyarch shipfiles/venv.zip \
      -pyclientexec venv.zip/venv/bin/python3 \
      -pyexec venv.zip/venv/bin/python3 \
      -py shipfiles/word_count.py

注意 它假定执行作业所需的 Python 依赖项已放置在 /path/to/shipfiles 目录中。例如，对于上面的示例，它应该包含 venv.zip 和 word_count.py。
注意：当它在 YARN 应用程序模式下执行 JobManager 上的作业时，-pyarch 和 -py 中指定的路径是相对于 shipfile 的路径，shipfile 是已交付文件的目录名称。
注意：通过 -pyarch 指定的存档文件将通过 Blob 服务器分发到任务管理器，其中文件大小限制为 2 GB。如果归档文件的大小超过 2 GB，则可以将其上载到分布式文件系统，然后使用命令行选项 -pyarch 中的路径。

• 在具有集群ID的本机Kubernetes集群上运行PyFlink应用程序，它需要一个安装了PyFlink的Docker映像，请参阅在docker中启用PyFlink：

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

创建一个保存点

可以创建保存点以保存作业所处的当前状态。所需要的只是工作 ID：

./bin/flink savepoint $JOB_ID  /tmp/flink-savepoints 

savepoint 文件夹是可选的，如果未设置 state.savepoints.dir，则需要指定该文件夹。
最后，您可以选择提供保存点的二进制格式。
保存点的路径可以在以后用于重新启动 Flink 作业。

删除保存点

保存点操作还可用于删除保存点。–dispose 需要添加相应的保存点路径：

./bin/flink savepoint \ 
      --dispose \
      /tmp/flink-savepoints/savepoint-cca7bc-bb1e257f0dab \ 
      $JOB_ID

如果使用自定义状态实例（例如自定义还原状态或 RocksDB 状态），则必须指定用于触发保存点的程序 JAR 的路径。否则，您将遇到一个 ClassNotFoundException：

./bin/flink savepoint \
      --dispose <savepointPath> \ 
      --jarfile <jarFile>

终止作业

正常停止作业以创建最终保存点

停止作业的另一个操作是 stop。这是一种在停止从源流向接收器时停止正在运行的流式处理作业的更优雅方式。当用户请求停止作业时，将请求所有源发送将触发保存点的最后一个检查点屏障，并且在该保存点成功完成后，它们将通过调用 cancel（） 方法完成。

./bin/flink stop \
      --savepointPath /tmp/flink-savepoints \
      $JOB_ID

如果未设置 state.savepoints.dir，我们必须使用 --savepointPath 指定 savepoint 文件夹。
如果指定了 --drain 标志，则在最后一个检查点屏障之前将发出MAX_WATERMARK。这将使所有已注册的事件时间计时器都处于触发状态，从而清除正在等待特定水印的任何状态，例如窗口。作业将继续运行，直到所有源正确关闭。这允许作业完成对所有动态数据的处理，这可以生成一些记录，以便在停止时获取的保存点之后进行处理。
如果要永久终止作业，请使用 --drain 标志。如果要在以后的某个时间点恢复作业，请不要排空管道，因为在恢复作业时可能会导致不正确的结果。

非优雅的停止作业

取消作业可以通过取消操作实现：

./bin/flink cancel $JOB_ID

相应作业的状态将从“正在运行”转换为“已取消”。任何计算都将停止。
–withSavepoint 标志允许在作业取消过程中创建保存点。此功能已弃用。请改用停止操作。

从保存点启动作业

可以使用运行（和运行应用程序）操作从保存点启动作业。

./bin/flink run \
      --detached \ 
      --fromSavepoint /tmp/flink-savepoints/savepoint-cca7bc-bb1e257f0dab \
      ./examples/streaming/StateMachineExample.jar

查看该命令如何等于初始 run 命令，但 --fromSavepoint 参数除外，该参数用于引用以前停止的作业的状态。将生成一个新的作业 ID，可用于维护作业。
默认情况下，我们尝试将整个保存点状态与正在提交的作业进行匹配。如果要允许跳过无法使用新作业还原的保存点状态，可以设置 --allowNonRestoredState 标志。如果从程序中删除了在触发保存点时属于程序一部分的算子，并且您仍希望使用保存点，则需要允许此操作。

./bin/flink run \
      --fromSavepoint <savepointPath> \
      --allowNonRestoredState ...

如果您的程序删除了作为保存点一部分的算子，这将非常有用。
您还可以选择应用于保存点的恢复模式。该模式控制谁将取得指定保存点文件的所有权。

希望对正在查看文章的您有所帮助，记得关注、评论、收藏，谢谢您