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

本文帶您快速體驗Flink Python流作業和批作業的建立、部署和啟動，以瞭解Realtime ComputeFlink版Python作業的操作流程。

前提條件

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

步驟一：下載Python測試檔案

本文為您提供了測試Python檔案和輸入資料檔案，您可以直接單擊下載待後續步驟使用。

您可以單擊下載以下任意Python測試作業。
- 單擊word_count_streaming.py，下載Python流作業測試Python檔案。
- 單擊word_count_batch.py，下載Python批作業測試Python檔案。
單擊Shakespeare，下載輸入資料檔案Shakespeare。

Flink控制台不提供Python包的開發平台，因此您需要線上下完成Python包的開發。Python作業的開發方法，調試及連接器的使用等詳情請參見Python作業開發。

步驟二：上傳Python檔案

登入Realtime Compute控制台。
單擊目標工作空間操作列下的控制台。
在左側導覽列，單擊資源管理。
單擊上傳資源，選擇測試Python檔案和資料檔案。
上傳後，檔案實際會被存放至名稱為您開通Flink工作空間時綁定的OSS Bucket下的artifacts目錄。上傳後，檔案路徑格式為oss://<您綁定的OSS Bucket名稱>/artifacts/namespaces/<專案空間名稱>目錄下。

說明

在Python作業中使用其他依賴（例如自訂的Python虛擬環境、第三方Python包、JAR包和資料檔案等）的方法請參見使用Python依賴。

步驟三：建立作業

流作業

登入Realtime Compute控制台。
單擊目標工作空間操作列下的控制台。
在作業營運介面，單擊部署作業，選擇Python作業。

填寫部署資訊。

參數	說明	樣本
部署模式	請選擇部署為流模式。	流模式
部署名稱	填寫對應的Python作業名稱。	flink-streaming-test-python
引擎版本	當前作業使用的Flink的引擎版本。引擎版本號碼含義、版本對應關係和生命週期重要時間點詳情請參見引擎版本介紹。	vvr-8.0.7-flink-1.17
Python檔案地址	單擊word_count_streaming.py下載測試Python檔案後，再單擊右側表徵圖選擇檔案，上傳Python檔案。	-
Entry Module	程式的入口類。如果Python作業檔案為.py檔案，則該項不需要填寫。如果Python作業檔案為.zip檔案，則需要在此處輸入您的Entry Module，例如word_count。	無需填寫
Entry Point Main Arguments	填寫輸入資料檔案的存放路徑。	`--input oss://<您綁定的OSS Bucket名稱>/artifacts/namespaces/<專案空間名稱>/Shakespeare`
Python Libraries	第三方Python包。第三方Python包會被添加到Python worker進程的PYTHONPATH中，從而在Python自訂函數中可以直接存取。如何使用第三方Python包，詳情請參見使用第三方Python包。	無需填寫
Python Archives	封存檔案，Python Archives詳情請參見使用自訂的Python虛擬環境和使用資料檔案。	無需填寫
附加依賴檔案	填寫目標附加依賴檔案的OSS路徑或者URL。	無需填寫
部署目標	在下拉式清單中，選擇目標資源隊列或者Session叢集（請勿生產使用）。詳情請參見管理資源隊列和步驟一：建立Session叢集。說明部署到Session叢集的作業不支援顯示資料曲線、配置監控警示和開啟自動調優功能。請勿將Session叢集用於正式生產環境，Session叢集可以作為開發測試環境。詳情請參見作業調試。	default-queue
備忘	可選，填寫備忘資訊。	無需填寫
作業標籤	配置作業標籤後，您可以在作業營運頁面根據標籤名和標籤值快速過濾找到目標作業。您最多可以建立3組作業標籤。	無需填寫
更多設定	開啟該開關後，您需要配置以下資訊： Kerberos叢集：單擊左側下拉式清單選擇您已建立的Kerberos叢集，Kerberos叢集建立操作詳情請參見註冊Hive Kerberos叢集。 principal：Kerberos principal又稱為主體，主體可以是使用者或服務，用於在Kerberos加密系統中標記一個唯一的身份。	無需填寫

單擊部署。

批作業

登入Realtime Compute控制台。
單擊目標工作空間操作列下的控制台。
在作業營運介面，單擊部署作業，選擇Python作業。

填寫部署資訊。

參數	說明	樣本
部署模式	請選擇部署為批模式。	批模式
部署名稱	填寫對應的Python作業名稱。	flink-batch-test-python
引擎版本	當前作業使用的Flink的引擎版本。引擎版本號碼含義、版本對應關係和生命週期重要時間點詳情請參見引擎版本介紹。	vvr-8.0.7-flink-1.17
Python檔案地址	單擊word_count_batch.py下載測試Python檔案後，再單擊右側表徵圖選擇檔案，上傳Python檔案。	-
Entry Module	程式的入口類。如果Python作業檔案為.py檔案，則該項不需要填寫。如果Python作業檔案為.zip檔案，則需要在此處輸入您的Entry Module，例如word_count。	無需填寫
Entry Point Main Arguments	填寫輸入以及輸出資料檔案的OSS路徑。說明本樣本中輸入資料檔案、輸出檔案和測試Python存放路徑一致。統一放在OSS控制台，名稱為flink-test-oss的Bucket下。本樣本中，將計算結果寫入到OSS指定目錄為例，為您展示如何配置該參數。在此您僅需要指定結果資料輸出檔案路徑和名稱，無需提前在指定目錄建立。	`--input oss://<您綁定的OSS Bucket名稱>/artifacts/namespaces/<專案空間名稱>/Shakespeare` `--output oss://<您綁定的OSS Bucket名稱>/artifacts/namespaces/<專案空間名稱>/python-batch-quickstart-test-output.txt`
Python Libraries	第三方Python包。第三方Python包會被添加到Python worker進程的Python路徑中，從而在Python自訂函數中可以直接存取。如何使用第三方Python包，詳情請參見使用第三方Python包。	無需填寫
Python Archives	封存檔案，Python Archives詳情請參見使用自訂的Python虛擬環境和使用資料檔案。	無需填寫
附加依賴檔案	填寫目標附加依賴檔案的OSS路徑或者URL。	無需填寫
部署目標	在下拉式清單中，選擇目標資源隊列或者Session叢集（請勿生產使用）。詳情請參見管理資源隊列和步驟一：建立Session叢集。說明部署到Session叢集的作業不支援顯示資料曲線、配置監控警示和開啟自動調優功能。請勿將Session叢集用於正式生產環境，Session叢集可以作為開發測試環境。詳情請參見作業調試。	default-queue
備忘	可選，填寫備忘資訊。	無需填寫
作業標籤	配置作業標籤後，您可以在作業營運頁面根據標籤名和標籤值快速過濾找到目標作業。您最多建立3組作業標籤。	無需填寫
更多設定	開啟該開關後，您需要配置以下資訊： Kerberos叢集：單擊左側下拉式清單選擇您已建立的Kerberos叢集，Kerberos叢集建立操作詳情請參見註冊Hive Kerberos叢集。 principal：Kerberos principal又稱為主體，主體可以是使用者或服務，用於在Kerberos加密系統中標記一個唯一的身份。	無需填寫

單擊部署。

步驟四：啟動Python作業

在作業營運頁面，單擊目標作業名稱操作列中的啟動。
單擊啟動後會彈出作業啟動對話方塊，配置對話方塊詳情請參見作業啟動。
單擊對話方塊中的啟動。
單擊啟動後，您可以看到作業狀態變為運行中或已完成，則代表作業運行正常。
重要
- 如果您使用文檔Python測試檔案，作業最終運行狀態是完成狀態。
- 如果您需要啟動批作業，則需要在作業營運頁面，將作業類型切換為批作業，才可以看到您上線的批作業。系統預設展示的作業為流作業。

步驟五：查看Flink計算結果

查看Python流作業計算結果：
在TaskManager中以.out結尾的記錄檔中，搜尋shakespeare查看Flink計算結果。
重要
如果您使用的是Python測試檔案，流作業變為已完成狀態時會刪除作業結果，故流作業狀態為運行中才能查看計算結果。
查看Python批作業計算結果：
在OSS管理主控台，oss://<您綁定的OSS Bucket名稱>/artifacts/namespaces/<專案空間名稱>/python-batch-quickstart-test-output.txt目錄，單擊名稱是作業的啟動日期和時間的檔案夾，然後單擊目標檔案名，在彈出的面板上單擊下載。
批作業結果為ext檔案，下載後可以用記事本或者Microsoft Office Word開啟查看結果，計算結果如下圖所示。

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

如果您對作業進行了修改且希望修改生效，則需要將運行中或已完成的作業停止再啟動。另外，如果作業無法複用State，希望作業全新啟動時，也需要停止後再啟動作業。作業停止詳情請參見作業停止。

Realtime Compute for Apache Flink：Flink Python作業快速入門