本文基于flink 1.13.x yarn 3.1.x撰寫并驗證
1 前提
-
安裝yarn集群并啟動
-
配置HADOOP_CLASSPATH環境變數
export HADOOP_CLASSPATH=`hadoop classpath`
注意:默認情況下,所有必需的Hadoop組態檔都是通過HADOOP_CLASSPATH環境變數從類路徑加載的,
2 Flink在YARN上支持的部署方式
對于生產使用,建議Per-job or Application Mode部署Flink應用程式,因為這些模式為應用程式提供了更好的隔離,
詳見參見: flink on yarn
2.1 Application Mode
Application Mode將在YARN上啟動一個Flink集群,其中應用程式jar的main()方法將在YARN中的JobManager上執行,
應用程式一完成,集群就會關閉,
可以通過yarn application -kill 或取消Flink作業手動停止集群,
#提交任務
./bin/flink run-application -t yarn-application ./examples/batch/WordCount.jar --input hdfs://pci01:8020/tmp/README.txt
./bin/flink run-application -t yarn-application ./examples/streaming/TopSpeedWindowing.jar -d
#列出集群上正在運行的作業,列出jobId、jobName
./bin/flink list -t yarn-application -Dyarn.application.id=application_1626944535001_1008
#取消任務: jobId
#請注意,取消應用程式集群上的作業將停止該集群,
./bin/flink cancel -t yarn-application -Dyarn.application.id=application_1626944535001_1008 825e0311eef66f586d0977a7c20a432d
為了釋放Application Mode的全部潛力,請考慮與yarn.provided.lib.dirs配置項一起使用
將flink的依賴jar、應用程式jar上傳到集群中所有節點都可以訪問的位置,
下述操作將使作業提交變得更加輕量級,因為所需的Flink jar和應用程式jar將由指定的遠程位置提取,而不是由客戶機發送到集群,
#將flink lib及應用的jar上傳到hdfs
hdfs dfs -mkdir -p hdfs://pci01:8020/jars/flink
hdfs dfs -copyFromLocal lib/*.jar hdfs://pci01:8020/jars/flink/
hdfs dfs -mkdir -p hdfs://pci01:8020/jars/apps
hdfs dfs -copyFromLocal ./examples/streaming/TopSpeedWindowing.jar hdfs://pci01:8020/jars/apps/
hdfs dfs -copyFromLocal ./examples/batch/WordCount.jar hdfs://pci01:8020/jars/apps/
#運行TopSpeedWindowing
./bin/flink run-application -t yarn-application -Dyarn.provided.lib.dirs="hdfs://pci01:8020/jars/flink/" hdfs://pci01:8020/jars/apps/TopSpeedWindowing.jar
#運行wordcount示例
./bin/flink run-application -t yarn-application -Dyarn.provided.lib.dirs="hdfs://pci01:8020/jars/flink/" hdfs://pci01:8020/jars/apps/WordCount.jar --input hdfs://pci01:8020/tmp/README.txt --output hdfs://pci01:8020/tmp/wordcount-result10.txt
hdfs dfs -cat hdfs://pci01:8020/tmp/wordcount-result10.tx
2.2 Per-Job Cluster Mode
Per-Job Cluster Mode將在YARN上啟動一個Flink集群,然后在本地運行提供的應用程式jar,最后將JobGraph提交給YARN上的JobManager,如果傳遞–detached引數,客戶端將在提交被接受后停止,
一個任務會對應一個Job,每提交一個作業會根據自身的情況,都會單獨向yarn申請資源,直到作業執行完成,一個作業的失敗與否并不會影響下一個作業的正常提交和運行,獨享Dispatcher和ResourceManager,按需接受資源申請;適合規模大長時間運行的作業,
一旦作業停止,Flink集群就會停止,
#提交任務
./bin/flink run -t yarn-per-job --detached ./examples/streaming/TopSpeedWindowing.jar
#列出集群上正在運行的作業, 列出jobId、jobName
./bin/flink list -t yarn-per-job -Dyarn.application.id=application_1626944535001_1015
#取消任務: jobId
#請注意,取消應用程式集群上的作業將停止該集群,
./bin/flink cancel -t yarn-per-job -Dyarn.application.id=application_1626944535001_1015 3386aec1fb6fba288c13a17b5cb6730b
2.3 Session Mode
Session-Cluster模式需要先啟動集群,然后再提交作業,接著會向yarn申請一塊空間后,資源永遠保持不變,如果資源滿了,下一個作業就無法提交,只能等到yarn中的其中一個作業執行完成后,釋放了資源,下個作業才會正常提交,所有作業共享Dispatcher和ResourceManager;共享資源;適合規模小執行時間短的作業,
Session-Cluster的資源在啟動集群時就定義完成,后續所有作業的提交都共享該資源,作業可能會互相影響,因此比較適合小規模短時間運行的作業
Session Mode將在/tmp/.yarn-properties-目錄下創建一個隱藏的YARN properties檔案,它將在提交作業時用于集群發現,并在命令列界面中顯示,有如下兩種操作模式:
-
attached mode (default): yarn-session.sh 客戶端將Flink集群提交給YARN,但是客戶端繼續運行,跟蹤集群的狀態,如果集群失敗,客戶端將顯示錯誤,如果客戶端被終止,它也會發出關閉集群的信號,
-
detached mode (
-dor--detached): 將Flink集群提交給YARN,客戶端回傳,需要通過以下命令停止#優雅停止 echo "stop" | ./bin/yarn-session.sh -id application_1626944535001_0984 #停止 yarn application -kill application_1626944535001_0984
除了通過conf/flink-conf.yaml檔案傳遞配置,也可以使用-Dkey=value引數在提交時將任何配置傳遞./bin/yarn-session.sh客戶端
2.3.1 在YARN上啟動一個長期的Flink集群
-s引數表示每個Task Manager上可用的處理槽(processing slot)數量,建議把槽數量設定成每個機器處理器的個數, 一旦會話被啟動,就可以使用./bin/flink工具提交任務到集群上,
cd flink/flink-1.13.2
./bin/yarn-session.sh -jm 2048 -tm 4096 -s 16 -d
#優雅停止
echo "stop" | ./bin/yarn-session.sh -id application_1626944535001_0984
#停止
yarn application -kill application_1626944535001_0984
# 查看可以獲得的命令
echo "help" | ./bin/yarn-session.sh -id application_1626944535001_0984
# 查看運行的application
yarn top
2.3.1.1 查看啟動的flink集群
在 Yarn Cluster頁面 http://172.25..:8088/cluster/apps/RUNNING,查看 name為Flink session cluster的application, 點擊Tracking UI中的ApplicationMaster鏈接,跳轉到Flink dashboard頁面


在yarn-session.sh成功運行后,日志顯示如下:
在日志的末尾顯示了 flink dashboard地址 及 applicationId

2.3.1.2 查看配置資訊
more /tmp/.yarn-properties-root
顯示內容如下:
#Generated YARN properties file
#Fri Sep 10 11:43:19 CST 2021
dynamicPropertiesString=
applicationID=application_1626944535001_1021
2.3.2 yarn-session.sh腳本引數
| 引數 | 名稱 | 解釋 |
|---|---|---|
| -at,–applicationType | Set a custom application type for the application on YARN | |
| -D <property=value> | use value for given property | |
| -d,–detached | If present, runs the job in detached mode | |
| -h,–help | Help for the Yarn session CLI. | |
| -id,–applicationId | Attach to running YARN session | |
| -j,–jar | Path to Flink jar file | |
| -jm,–jobManagerMemory | Memory for JobManager Container with optional unit (default: MB) | |
| -m,–jobmanager | Set to yarn-cluster to use YARN execution mode. | |
| -nl,–nodeLabel | Specify YARN node label for the YARN application | |
| -nm,–name | Set a custom name for the application on YARN | |
| -q,–query | Display available YARN resources (memory, cores) | |
| -qu,–queue | Specify YARN queue. | |
| -s,–slots | Number of slots per TaskManager | |
| -t,–ship | Ship files in the specified directory (t for transfer) | |
| -tm,–taskManagerMemory Memory per TaskManager Container with optional unit (default: MB) | ||
| -yd,–yarndetached | If present, runs the job in detached mode (deprecated; use non-YARN specific option instead) | |
| -z,–zookeeperNamespace | Namespace to create the Zookeeper sub-paths for high availability mode |
2.3.3 提交任務
./bin/flink run ./examples/streaming/TopSpeedWindowing.jar
./bin/flink run ./examples/batch/WordCount.jar --input hdfs://pci01:8020/tmp/README.txt --output hdfs://pci01:8020/tmp/wordcount-result20.txt
./bin/flink run -t yarn-session \
-Dyarn.application.id=application_1626944535001_1021 \
./examples/batch/WordCount.jar --input hdfs://pci01:8020/tmp/README.txt \
--output hdfs://pci01:8020/tmp/wordcount-result21.txt
3 flink on yarn配置
flink on yarn配置
以下配置引數是由Flink on YARN管理的,因為它們可能會在運行時被框架覆寫:
- jobmanager.rpc.address: 通過Flink on YARN上動態設定為JobManager容器的地址
- io.tmp.dirs:如果沒有設定,Flink將設定YARN定義的臨時目錄
- high-availability.cluster-id : 自動生成的ID,用于區分HA服務中的多個集群
如果需要將額外的Hadoop組態檔傳遞給Flink,可以通過HADOOP_CONF_DIR環境變數來實作,該變數接受一個包含Hadoop組態檔的目錄名,默認情況下,所有必需的Hadoop組態檔都是通過HADOOP_CLASSPATH環境變數從類路徑加載的,
4 Resource Allocation Behavior 資源配置行為
如果一個運行在YARN上的JobManager不能使用現有資源運行所有提交的作業,那么它將請求額外的taskManager,特別是在Session Mode下運行時,如果需要,JobManager會在提交額外的作業時分配額外的taskManager,未使用的taskManager在超時后再次釋放,
YARN實作將尊重JobManager和TaskManager行程的記憶體配置,默認情況下,報告的VCores數量等于每個TaskManager配置的槽位數量,yarn.containers.vcores允許使用自定義值覆寫vcores的數量,為了讓這個引數生效,您應該在YARN集群中啟用CPU scheduling ,
失敗的容器(包括JobManager)將被YARN替換,JobManager容器重啟的最大次數是通過 yarn.application-attempts[默認是1]配置的,當所有的嘗試都耗盡時,YARN應用程式將失敗,
5 User jars & Classpath
默認情況下,Flink在運行單個作業時將用戶jar包含到系統類路徑【system classpath】中,這種行為可以通過yarn.per-job-cluster.include-user-jar 引數進行控制,
當將此設定為DISABLED時,表示用戶jar被排除在系統類路徑之外,
用戶jar在類路徑中的位置可以通過將引數設定為以下引數之一來控制:
ORDER: (default) Adds the jar to the system classpath based on the lexicographic order.FIRST: Adds the jar to the beginning of the system classpath.LAST: Adds the jar to the end of the system classpath.
6 三種模式的區別
詳見 deployment
主要區別點如下:
- 集群生命周期和資源隔離保證
- 應用程式的main()方法是在client上執行還是在cluster上執行,
- 在session mode中,如果其中一個作業行為不當或關閉了一個TaskManager,那么在該TaskManager上運行的所有作業都將受到該失敗的影響
應用程式的main方法包括的程序如下:
- 在本地下載應用程式的依賴項,
- 執行main()來提取Flink的運行時可以理解的應用程式的表示(即JobGraph),
- 并將依賴項和JobGraph發送到集群
如果main在client執行,則client會成為一個沉重的資源消耗者,因為它可能需要大量的網路帶寬來下載依賴項并將二進制檔案發送到集群,并需要CPU周期來執行main(),當client 服務器在用戶之間共享時,這個問題會更加明顯,

Application Mode 與 Per-Job Mode的區別
- Application Mode的main代碼在cluster中執行,Per-Job Mode在client中執行
- 與Per-Job模式相比,Application模式允許提交由多個作業Job組成的應用程式,
轉載請註明出處,本文鏈接:https://www.uj5u.com/qita/299486.html
標籤:其他
上一篇:Linux系統安裝Hive客戶端
