Flink Python快速入門 - Realtime Compute for Apache Flink

前提條件

如果您使用RAM使用者或RAM角色等身份訪問，需要確認已具有Flink控制台相關許可權，詳情請參見許可權管理。
已建立Flink工作空間，詳情請參見開通Realtime ComputeFlink版。

步驟一：準備Python代碼檔案

Realtime Compute管理主控台不提供Python的開發環境，您需要在本地完成作業開發。有關作業調試和連接器的使用，詳情請參見Python作業開發。

重要

本地開發依賴的Flink版本需確保與後續步驟三：部署Python作業選擇的引擎版本保持一致，在Python作業中使用其他依賴（自訂的Python虛擬環境、第三方Python包、JAR包和資料檔案等）的方法請參見使用Python依賴。

為了協助您快速熟悉Flink Python作業操作，本文已為您提供統計單詞出現頻率的測試Python檔案和資料文本，您可以直接下載待後續步驟使用。

根據需要下載Python測試作業。
- 流作業：word_count_streaming.py。
- 批作業：word_count_batch.py。

單擊Shakespeare，下載資料文本Shakespeare。

步驟二：上傳Python檔案和資料檔案

登入Realtime Compute控制台。
單擊目標工作空間操作列下的控制台。
在左側導覽列，單擊檔案管理。
單擊上傳資源，上傳要部署的Python檔案和資料檔案。

此處上傳步驟一下載的Python測試檔案和資料檔案，檔案儲存體路徑詳情請參見檔案管理。

步驟三：部署Python作業

流作業

在營運中心 > 作業營運介面，單擊部署作業 > Python作業。

填寫部署資訊。

參數	說明	樣本
部署模式	請選擇部署為流模式。	流模式
部署名稱	填寫對應的Python作業名稱。	flink-streaming-test-python
引擎版本	當前作業使用的Flink引擎版本。建議使用帶有推薦、穩定標籤的版本，這些版本具有更高的可靠性和效能表現，詳情請參見功能發布記錄和引擎版本介紹。	vvr-8.0.9-flink-1.17
Python檔案地址	單擊word_count_streaming.py下載測試Python檔案後，再單擊右側表徵圖選擇檔案，上傳Python檔案。如果檔案管理中已存在該檔案，則無需重複上傳，直接選擇目標檔案即可。	-
Entry Module	程式的入口類。如果Python作業檔案為.py檔案，則該項不需要填寫。如果Python作業檔案為.zip檔案，則需要在此處輸入您的Entry Module，例如word_count。	無需填寫
Entry Point Main Arguments	填寫傳入參數資訊，在主方法裡面調用該參數。本文填寫輸入資料檔案Shakespeare的存放路徑。	`--input oss://<您綁定的OSS Bucket名稱>/artifacts/namespaces/<專案空間名稱>/Shakespeare` 您可以直接在檔案管理中複製Shakespeare檔案的完整路徑。
部署目標	在下拉式清單中，選擇目標資源隊列或者Session叢集（請勿生產使用）。詳情請參見管理資源隊列和建立Session叢集。重要部署到Session叢集的作業不支援顯示監控警示（或資料曲線）、配置監控警示和開啟自動調優功能。請勿將Session叢集用於正式生產環境，Session叢集可以作為開發測試環境。詳情請參見作業調試。	default-queue

更多配置參數詳情請參見部署作業。

單擊部署。

批作業

在營運中心 > 作業營運介面，單擊部署作業，選擇Python作業。

填寫部署資訊。

參數	說明	樣本
部署模式	請選擇部署為批模式。	批模式
部署名稱	填寫對應的Python作業名稱。	flink-batch-test-python
引擎版本	當前作業使用的Flink引擎版本。建議使用帶有推薦、穩定標籤的版本，這些版本具有更高的可靠性和效能表現，詳情請參見功能發布記錄和引擎版本介紹。	vvr-8.0.9-flink-1.17
Python檔案地址	單擊word_count_batch.py下載測試Python檔案後，再單擊右側表徵圖選擇檔案，上傳Python檔案。	-
Entry Module	程式的入口類。如果Python作業檔案為.py檔案，則該項不需要填寫。如果Python作業檔案為.zip檔案，則需要在此處輸入您的Entry Module，例如word_count。	無需填寫
Entry Point Main Arguments	填寫傳入參數資訊，在主方法裡面調用該參數。本文填寫輸入資料檔案Shakespeare和輸出資料目錄batch-quickstart-test-output的存放路徑。說明您只需指定輸出目錄的路徑名稱，無需提前在儲存服務中建立輸出目錄，輸出目錄的父目錄路徑與輸入檔案保持一致即可。	`--input oss://<您綁定的OSS Bucket名稱>/artifacts/namespaces/<專案空間名稱>/Shakespeare` `--output oss://<您綁定的OSS Bucket名稱>/artifacts/namespaces/<專案空間名稱>/python-batch-quickstart-test-output` 您可以直接在檔案管理中複製Shakespeare檔案的完整路徑。
部署目標	在下拉式清單中，選擇目標資源隊列或者Session叢集（請勿生產使用）。詳情請參見管理資源隊列和建立Session叢集。重要部署到Session叢集的作業不支援顯示監控警示、配置監控警示和開啟自動調優功能。請勿將Session叢集用於正式生產環境，Session叢集可以作為開發測試環境。詳情請參見作業調試。	default-queue

更多配置參數詳情請參見部署作業。

單擊部署。

步驟四：啟動Python作業並查看Flink計算結果

流作業

在營運中心 > 作業營運頁面，單擊目標作業名稱操作列中的啟動。
選擇無狀態啟動，單擊啟動，作業啟動詳情請參見作業啟動。

單擊啟動後，作業狀態變為運行中或已完成，則代表作業運行正常。如果您部署本文檔Python測試檔案，作業最終運行狀態是已完成狀態。
作業狀態變為運行中後，查看流作業樣本的計算結果。

重要
如果您部署的是本文Python測試檔案，流作業變為已完成狀態時會刪除作業結果，故流作業狀態為運行中才能看到計算結果。

在TaskManager中以.out結尾的記錄檔中，搜尋shakespeare查看Flink計算結果。

具體操作路徑：單擊作業日誌頁簽，選擇運行 Task Managers標籤頁，進入對應 TaskManager 後單擊日誌列表子頁簽，開啟 flink.out 檔案，在右上方搜尋方塊中輸入 shakespeare 即可定位到詞頻統計結果，例如 (shakespeare,1)。

批作業

在營運中心 > 作業營運頁面，單擊目標作業中的啟動。

在篩選欄的作業類型下拉框中選擇批作業，找到目標作業後單擊其右側的啟動。
在作業啟動對話方塊中，單擊啟動，作業啟動詳情請參見作業啟動。
作業狀態變為已完成後，查看批作業樣本的計算結果。

登入OSS管理主控台，在oss://<您綁定的OSS Bucket名稱>/artifacts/namespaces/<專案空間名稱>/python-batch-quickstart-test-output目錄，單擊名稱是作業的啟動日期和時間的檔案夾，然後單擊目標檔案名，在彈出的面板上單擊下載。
批作業結果為ext檔案，下載後可以用記事本或者Microsoft Office Word開啟查看結果，計算結果如下所示。
```
(As,40)
(At,5)
(Ay,1)
(Be,9)
(By,14)
(Do,4)
(He,7)
(I,,4)
(If,34)
(In,36)
(Is,10)
(It,6)
```

（可選）步驟五：停止作業

如果您對作業進行了修改（例如更改代碼、增刪改WITH參數、更改作業版本等），且希望修改生效，則需要重新部署作業，然後停止再啟動。另外，如果作業無法複用State，希望作業全新啟動時，或者更新非動態生效的參數配置時，也需要停止後再啟動作業。作業停止詳情請參見作業停止。

Realtime Compute for Apache Flink：Flink Python作業