歡迎訪問我的GitHub
https://github.com/zq2599/blog_demos
內容：所有原創文章分類匯總及配套原始碼，涉及Java、Docker、Kubernetes、DevOPS等；
本文是《Flink on Yarn三部曲》系列的終篇，先簡單回顧前面的內容：
- 《Flink on Yarn三部曲之一：準備作業》：準備好機器、腳本、安裝包；
- 《Flink on Yarn三部曲之二：部署和設定》：完成CDH和Flink部署，并在管理頁面做好相關的設定；
現在Flink、Yarn、HDFS都就緒了，接下來實踐提交Flink任務到Yarn執行；
全文鏈接
- 《Flink on Yarn三部曲之一：準備作業》
- 《Flink on Yarn三部曲之二：部署和設定》
- 《Flink on Yarn三部曲之三：提交Flink任務》
兩種Flink on YARN模式
實踐之前，對Flink on YARN先簡單了解一下，如下圖所示，Flink on Yarn在使用的時候分為兩種模式，Job Mode和Session Mode：
Session Mode：在YARN中提前初始化一個Flink集群，以后所有Flink任務都提交到這個集群，如下圖：
Job Mode：每次提交Flink任務都會創建一個專用的Flink集群，任務完成后資源釋放，如下圖：
接下來分別實戰這兩種模式；
準備實戰用的資料(CDH服務器)
接下來提交的Flink任務是經典的WordCount，先在HDFS中準備一份文本檔案，后面提交的Flink任務都會讀取這個檔案，統計里面每個單詞的數字，準備文本的步驟如下：
- SSH登錄CDH服務器；
- 切換到hdfs賬號：su - hdfs
- 下載實戰用的txt檔案：
wget https://github.com/zq2599/blog_demos/blob/master/files/GoneWiththeWind.txt
- 創建hdfs檔案夾：hdfs dfs -mkdir /input
- 將文本檔案上傳到/input目錄：hdfs dfs -put ./GoneWiththeWind.txt /input
準備作業完成，可以提交任務試試了，
Session Mode實戰
- SSH登錄CDH服務器；
- 切換到hdfs賬號：su - hdfs
- 進入目錄：/opt/flink-1.7.2/
- 執行如下命令創建Flink集群，-n引數表示TaskManager的數量，-jm表示JobManager的記憶體大小，-tm表示每個TaskManager的記憶體大小：
./bin/yarn-session.sh -n 2 -jm 1024 -tm 1024
- 創建成功后，控制臺輸出如下圖，注意紅框中的提示，表明可以通過38301埠訪問Flink：
- 瀏覽器訪問CDH服務器的38301埠，可見Flink服務已經啟動：
- 瀏覽器訪問CDH服務器的8088埠，可見YARN的Application(即Flink集群)創建成功，如下圖，紅框中是任務ID，稍后結束Application的時候會用到此ID：
- 再開啟一個終端，SSH登錄CDH服務器，切換到hdfs賬號，進入目錄：/opt/flink-1.7.2
- 執行以下命令，就會提交一個Flink任務（安裝包自帶的WordCount例子），并指明將結果輸出到HDFS的wordcount-result.txt檔案中：
bin/flink run ./examples/batch/WordCount.jar \
-input hdfs://192.168.50.134:8020/input/GoneWiththeWind.txt \
-output hdfs://192.168.50.134:8020/wordcount-result.txt
- 執行完畢后，控制臺輸出如下：
- flink的WordCount任務結果保存在hdfs，我們將結果取出來看看：hdfs dfs -get /wordcount-result.txt
- vi打開wordcount-result.txt檔案，如下圖，可見任務執行成功，指定文本中的每個單詞數量都統計出來了：
- 瀏覽器訪問Flink頁面(CDH服務器的38301埠)，也能看到任務的詳細情況：
- 銷毀這個Flink集群的方法是在控制臺執行命令：yarn application -kill application_1580173588985_0002
Session Mode的實戰就完成了，接下來我們來嘗試Job Mode；
Job Mode
- 執行以下命令，創建一個Flink集群，該集群只用于執行引數中指定的任務(wordCount.jar)，結果輸出到hdfs的wordcount-result-1.txt檔案：
bin/flink run -m yarn-cluster \
-yn 2 \
-yjm 1024 \
-ytm 1024 \
./examples/batch/WordCount.jar \
-input hdfs://192.168.50.134:8020/input/GoneWiththeWind.txt \
-output hdfs://192.168.50.134:8020/wordcount-result-1.txt
- 控制臺輸出如下，表明任務執行完成：
- 如果您的記憶體和CPU核數充裕，可以立即執行以下命令再創建一個Flink集群，該集群只用于執行引數中指定的任務(wordCount.jar)，結果輸出到hdfs的wordcount-result-2.txt檔案：
bin/flink run -m yarn-cluster \
-yn 2 \
-yjm 1024 \
-ytm 1024 \
./examples/batch/WordCount.jar \
-input hdfs://192.168.50.134:8020/input/GoneWiththeWind.txt \
-output hdfs://192.168.50.134:8020/wordcount-result-2.txt
- 在YARN管理頁面可見任務已經結束：
- 執行命令hdfs dfs -ls /查看結果檔案，已經成功生成：
- 執行命令hdfs dfs -get /wordcount-result-1.txt下載結果檔案到本地，檢查資料正常；
- 至此，Flink on Yarn的部署、設定、提交都實踐完成，《Flink on Yarn三部曲》系列也結束了，如果您也在學習Flink，希望本文能夠給您一些參考，也建議您根據自身情況和需求，修改ansible腳本，搭建更適合自己的環境；
歡迎關注公眾號：程式員欣宸
微信搜索「程式員欣宸」，我是欣宸，期待與您一同暢游Java世界...
https://github.com/zq2599/blog_demos
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/183612.html
標籤：其他