全部产品
Search

搜索本产品

    开源大数据平台E-MapReduce：基础使用

    文档中心

    开源大数据平台E-MapReduce：基础使用

    更新时间：Aug 03, 2026

    本文为您介绍如何在E-MapReduce上提交Flink作业以及查看作业。

    背景信息

    Dataflow集群中的Flink服务是以YARN模式部署的，您可以通过SSH方式登录Dataflow集群，在命令行中进行Flink作业提交。

    基于YARN模式部署的Dataflow集群支持以Session模式、Per-Job Cluster模式和Application模式提交Flink作业。

    模式

    描述

    特点

    Session模式

    Session模式会根据您设置的资源参数创建一个Flink集群，所有作业都将被提交到这个集群上运行。该集群在作业运行结束之后不会自动释放。

    例如，某个作业发生异常，导致一个Task Manager关闭，则其他所有运行在该Task Manager上的作业都会失败。另外由于同一个集群中只有一个Job Manager，随着作业数量的增多，Job Manager的压力会相应增加。

    • 优点：提交作业时，资源分配导致的时间开销相比其他模式较小。

    • 缺点：由于所有作业都运行在该集群中，会存在对资源的竞争以及作业间的相互影响。

    根据以上特点，该模式适合部署需要较短启动时间且运行时间相对较短的作业。

    Per-Job Cluster模式

    当使用Per-Job Cluster模式时，每次提交一个Flink作业，YARN都会为这个作业新启动一个Flink集群，然后运行该作业。当作业运行结束或者被取消时，该作业所属的Flink集群也会被释放。

    • 优点：作业之间资源隔离，一个作业的异常行为不会影响到其他作业。

      因为每个作业都和一个Job Manager一一对应，因此不会出现一个Job Manager因为运行多个Job而导致负载过高的问题。

    • 缺点：每次运行一个作业都要启动一个专属Flink集群，启动作业的开销更大。

    根据以上特点，该模式通常适合运行时间较长的作业。

    Application模式

    当使用Application模式时，每次提交一个Flink Application（一个Application包含一个或多个作业），YARN都会为这个Application新启动一个Flink集群。当Application运行结束或者被取消时，该Application所属的Flink集群也会被释放。

    该模式与Per-Job模式不同的是，Application对应的JAR包中的main()方法会在集群中的Job Manager中被执行。

    如果提交的JAR包中包含多个作业，则这些作业都会在该Application所属的集群中执行。

    • 优点：可以减轻客户端提交作业时的负担。

    • 缺点：每次运行一个Flink Application都要启动一个专属Flink集群，启动Application的时间开销会更大。

    前提条件

    已创建Flink模式的Dataflow集群，详情请参见创建集群。

    提交并查看Flink作业

    说明

    本文使用Flink自身提供的TopSpeedWindowing示例进行介绍，该示例是一个会长时间运行的流作业。

    您可以根据需求，选择以下三种模式提交并查看作业：

    Session模式

    1. 通过SSH方式连接集群的Master节点，具体操作请参见登录集群Master节点。

    2. 执行以下命令，启动YARN Session。

      yarn-session.sh --detached

      执行成功后，系统会返回Application ID。例如，application_1750137174986_0001，后续将使用<application_XXXX_YY>表示。

      mr.aliyuncs.com:33879 of application 'application_1750137174986_0001'.
JobManager Web Interface: http://core-1-1.c-1f6ec9xxx.cn-hangzhou.emr.aliyuncs.com:33879
2025-06-17 13:19:20,152 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - The Flink YARN session cluster has been started in detached mode. In order to stop Flink gracefully, use the following command:
$ echo "stop" | ./bin/yarn-session.sh -id application_1750137174986_0001
If this should not be possible, then you can also kill Flink via YARN's web interface or via:
$ yarn application -kill application_1750137174986_0001
Note that killing Flink might not clean up all job artifacts and temporary files.

    3. 执行以下命令，提交作业。

      flink run --detached /opt/apps/FLINK/flink-current/examples/streaming/TopSpeedWindowing.jar

      提交成功后，系统会返回如下类似信息。

      其中3785db18d371326758d7843dd2a1****为该作业ID，后续将使用<jobId>表示。

    4. 执行以下命令，查看作业状态。

      flink list -t yarn-session -Dyarn.application.id=<application_XXXX_YY>

      返回如下类似信息。

      ------------------ Running/Restarting Jobs -------------------
16.06.2025 18:20:55 : 3785db18d371326758d7843dd2a1**** : CarTopSpeedWindowingExample (RUNNING)

      您也可以通过Web UI的方式查看作业状态，详情请参见通过Web UI查看作业状态。

    5. 执行以下命令，停止作业。

      flink cancel -t yarn-session -Dyarn.application.id=<application_XXXX_YY> <jobId>

    Per-Job Cluster模式

    1. 通过SSH方式连接集群的Master节点，具体操作请参见登录集群Master节点。

    2. 执行以下命令，提交作业。

      flink run -t yarn-per-job --detached /opt/apps/FLINK/flink-current/examples/streaming/TopSpeedWindowing.jar

      提交成功后，系统会返回如下类似信息。

      $ yarn application -kill application_1750125819948_0003
Note that killing Flink might not clean up all job artifacts and temporary files.
2025-06-17 10:44:46,268 INFO  org.apache.flink.yarn.YarnClusterDescriptor                 [] - Found Web Interface core-1-1.c-b9693c.xxx.cn-hangzhou.emr.aliyuncs.com:38037 of application 'application_1750125819948_0003'.
Job has been submitted with JobID 451aded93de19d6cd238ed3b466xxx
You have new mail in /var/spool/mail/root

      其中application_1750125819948_****为Application ID，后续将使用<application_XXXX_YY>表示；f5f980ac631192b02548235f1bbe****为该作业ID，后续将使用<jobId>表示。

    3. 您可以执行以下命令，查看作业状态。

      flink list -t yarn-per-job -Dyarn.application.id=<application_XXXX_YY>

      您也可以通过Web UI的方式查看作业状态，详情请参见通过Web UI查看作业状态。

    4. 执行以下命令，停止作业。

      flink cancel -t yarn-per-job -Dyarn.application.id=<application_XXXX_YY> <jobId>

    Application模式

    1. 通过SSH方式连接集群的Master节点，具体操作请参见登录集群Master节点。

    2. 执行以下命令，提交作业。

      flink run-application -t yarn-application /opt/apps/FLINK/flink-current/examples/streaming/TopSpeedWindowing.jar

      提交成功后，系统会返回如下类似信息。

      [root@master-1-1(172.17.xxx.xxx) ~]# flink run-application -t yarn-application /opt/apps/FLINK/flink-current/examples/streaming/TopSpeedWindowing.jar
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/opt/apps/FLINK/flink-1.17.2-1.0.10/lib/log4j-slf4j-impl-2.17.1.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/apps/HADOOP-COMMON/hadoop-3.2.1-1.3.2-alinux3/share/hadoop/common/lib/slf4j-log4j12-1.7.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.apache.logging.slf4j.Log4jLoggerFactory]
2025-06-17 10:57:05,106 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Found Yarn properties file under /tmp/.yarn-properties-root.
2025-06-17 10:57:05,106 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Found Yarn properties file under /tmp/.yarn-properties-root.
2025-06-17 10:57:05,233 WARN  org.apache.flink.yarn.configuration.YarnLogConfigUtil        [] - The configuration directory ('/etc/taihao-apps/flink-conf') already contains a LOG4J config file.If you want to use logback, then please delete or rename the log configuration file.
2025-06-17 10:57:05,453 INFO  org.apache.hadoop.yarn.client.RMProxy                       [] - Connecting to ResourceManager at master-1-1.c-b9693c1xxx.cn-hangzhou.emr.aliyuncs.com/172.17.xxx.xxx:8032
2025-06-17 10:57:05,604 INFO  org.apache.hadoop.yarn.client.AHSProxy                      [] - Connecting to Application History server at master-1-1.c-b9693xxx
3c131faf601f.cn-hangzhou.emr.aliyuncs.com/172.17.108.111:10200
2025-06-17 10:57:05,612 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
2025-06-17 10:57:05,724 INFO  org.apache.hadoop.conf.Configuration                        [] - found resource resource-types.xml at file:/etc/taihao-apps/hadoop-conf/resource-types.xml
2025-06-17 10:57:05,776 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - The configured JobManager memory is 1600 MB. YARN will allocate 1664 MB to make up an integer multiple of its minimum allocation memory (128 MB, configured via 'yarn.scheduler.minimum-allocation-mb'). The extra 64 MB may not be used by Flink.
2025-06-17 10:57:05,776 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - The configured TaskManager memory is 1728 MB. YARN will allocate 1792 MB to make up an integer multiple of its minimum allocation memory (128 MB, configured via 'yarn.scheduler.minimum-allocation-mb'). The extra 64 MB may not be used by Flink.
2025-06-17 10:57:05,776 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - Cluster specification: ClusterSpecification{masterMemoryMB=1600, taskManagerMemoryMB=1728, slotsPerTaskManager=1}
2025-06-17 10:57:10,219 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - Cannot use kerberos delegation token manager, no valid kerberos credentials provided.
2025-06-17 10:57:10,227 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - Submitting application master application_1750125819948_0004
2025-06-17 10:57:10,271 INFO  org.apache.hadoop.yarn.client.api.impl.YarnClientImpl        [] - Submitted application application_1750125819948_0004
2025-06-17 10:57:10,271 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - Waiting for the cluster to be allocated
2025-06-17 10:57:10,278 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - Deploying cluster, current state ACCEPTED
2025-06-17 10:57:17,825 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - YARN application has been deployed successfully.
2025-06-17 10:57:17,825 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - Found Web Interface core-1-1.c-b9693c1xxx.cn-hangzhou.emr.aliyuncs.com:42563 of application 'application_1750125819948_0004'.

      其中，application_1750125819948_0004为已提交的Flink作业的YARN Application ID，后续将使用<application_XXXX_YY>表示。

    3. 执行以下命令，查看作业状态。

      flink list -t yarn-application -Dyarn.application.id=<application_XXXX_YY>

      返回如下类似信息，其中4db32b5339e6d64de2a1096c4762****为该作业的<jobId>。

      ------------------ Running/Restarting Jobs -------------------
16.06.2025 18:20:55 : 4db32b5339e6d64de2a1096c4762**** : CarTopSpeedWindowingExample (RUNNING)

      您也可以通过Web UI的方式查看作业状态，详情请参见通过Web UI查看作业状态。

    4. 执行以下命令，停止作业。

      flink cancel -t yarn-application -Dyarn.application.id=<application_XXXX_YY> <jobId>

    指定作业配置

    Flink提供三种指定作业配置的方式：

    • 方式一：在作业代码中，指定配置项的值，详情请查看Flink配置。

    • 方式二：使用flink run命令提交作业时，通过-D指定配置项的值，例如flink run-application -t yarn-application -D state.backend=rocksdb... 。

    • 方式三：在/etc/taihao-apps/flink-conf/flink-conf.yaml配置文件中指定配置项的值。

    如果没有通过这三种方式指定，则使用默认值，配置参数详情请参见Apache Flink官网。

    通过Web UI查看作业状态

    1. 访问Web UI。

      1. 登录E-MapReduce控制台。

      2. 在左侧导航栏，选择EMR on ECS。

      3. 在顶部菜单栏处，根据实际情况选择地域和资源组。

      4. 在EMR on ECS页面，单击目标集群的集群ID。

      5. 单击上方的访问链接与端口页签。

      6. 在访问链接与端口页面，单击YARN UI所在行的链接。

        访问Web UI的详细信息，请参见通过控制台访问开源组件Web界面。

    2. 单击Application ID。

      在 Hadoop YARN ResourceManager 的 All Applications 页面，找到名称为 Flink per-job cluster 的应用，单击其对应的 Application ID（例如 application_1628232179762_0002）。

    3. 单击Tracking URL的链接。

      在 Application Overview 区域，Tracking URL 对应的链接显示为 ApplicationMaster。

      进入Apache Flink Dashboard页面，即可查看作业的状态。

    相关文档

    Flink on YARN的更多信息，请参见Apache Hadoop YARN。