全部產品
Search

搜尋本產品

    E-MapReduce：基礎使用

    文件中心

    E-MapReduce：基礎使用

    更新時間：Aug 04, 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。