為 Python 應用程序創建並運行阿帕奇 Flink 的託管服務

建立資料串流 (AWS CLI)

1. 若要建立第一個串流 (ExampleInputStream)，請使用下列 Amazon Kinesis create-stream AWS CLI 指令。

   $ aws kinesis create-stream \
   --stream-name ExampleInputStream \
   --shard-count 1 \
   --region us-east-1 

2. 若要建立應用程式用來寫入輸出的第二個串流，請執行相同的命令，將串流名稱變更為 ExampleOutputStream。

   $ aws kinesis create-stream \
   --stream-name ExampleOutputStream \
   --shard-count 1 \
   --region us-east-1 

1. 使用以下指令複製遠端儲存庫：

   git clone https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples.git

2. 導覽至 ./python/GettingStarted 目錄。

若要從命令列執行應用程式

1. 請確定獨立 Python 環境 (例如 Conda) 或您安裝 Python Flink 程式庫的 VirtualEnv 位置目前處於作用中狀態。

2. 確保您至mvn package少跑了一次。

3. 設定 IS_LOCAL = true 環境變數：

   $ export IS_LOCAL=true

4. 運行應用程序作為一個常規的 Python 腳本。

   $python main.py

若要從中執行應用程式 IDE

1. 配置您IDE使用以下配置運行main.py腳本：

   1. 使用獨立的 Python 環境，例如 Conda 或您安裝程式 PyFlink 庫的 VirtualEnv 位置。

   2. 使用 AWS 認證來存取輸入和輸出 Kinesis 資料串流。

   3. 設定 IS_LOCAL = true。

2. 設置運行配置的確切過程取決於您的IDE和變化。

3. 設置完畢後IDE，請運行 Python 腳本並使用應用程序運行IDE時提供的工具。

上傳應用程式程式碼JAR檔案

1. 在開啟 Amazon S3 主控台https://console.aws.amazon.com/s3/。

2. 選擇您先前為應用程式程式碼建立的值區。

3. 選擇上傳。

4. 選擇 Add files (新增檔案)。

5. 導覽至在上一個步驟中產生的 .zip 檔案：target/managed-flink-pyflink-getting-started-1.0.0.zip。

6. 選擇「上傳」而不變更任何其他設定。

1. 在 https://console.aws.amazon.com /flink 開啟適用於阿帕奇 Flink 的受管理服務

2. 確認已選取正確的區域：美國東部 (維吉尼亞北部) us-east-1。

3. 開啟右側功能表並選擇 Apache Flink 應用程式，然後選擇建立串流應用程式。或者，從初始頁面的 [開始使用] 區段中選擇 [建立串流應用程式]。

4. 在 [建立串流應用程式] 頁面上：

   • 對於選擇設定串流處理應用程式的方法，請選擇從頭開始建立。

   • 對於阿帕奇 Flink 配置，應用程序 Flink 版本，請選擇阿帕奇 F link 1.19。

   • 對於應用程序配置：

     • 在應用程式名稱中，輸入 MyApplication。

     • 對於 Description (說明)，輸入 My Python test app。

     • 在 [存取應用程式資源] 中，選擇 [建立/更新IAM角色運動分析 MyApplication--us- east-1 具有必要原則]。

   • 對於應用程式設定的範本：

     • 針對範本，選擇開發。

   • 選擇建立串流應用程式。

編輯IAM政策以新增 S3 儲存貯體許可

1. 在開啟IAM主控台https://console.aws.amazon.com/iam/。

2. 選擇政策。選擇主控台為您在上一節所建立的 kinesis-analytics-service-MyApplication-us-east-1 政策。

3. 選擇 [編輯]，然後選擇索JSON引標籤。

4. 將下列政策範例的反白部分新增至政策。取代範例帳戶 IDs (012345678901) 使用您的帳戶 ID。

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:s3:::my-bucket/kinesis-analytics-placeholder-s3-object"
               ]
           },
           {
               "Sid": "ListCloudwatchLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:us-east-1:012345678901:log-group:*"
               ]
           },
           {
               "Sid": "ListCloudwatchLogStreams",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogStreams"
               ],
               "Resource": [
                   "arn:aws:logs:us-east-1:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:*"
               ]
           },
           {
               "Sid": "PutCloudwatchLogs",
               "Effect": "Allow",
               "Action": [
                   "logs:PutLogEvents"
               ],
               "Resource": [
                   "arn:aws:logs:us-east-1:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:kinesis-analytics-log-stream"
               ]
           },
           {
               "Sid": "ReadInputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-east-1:012345678901:stream/ExampleInputStream"
           },
           {
               "Sid": "WriteOutputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-east-1:012345678901:stream/ExampleOutputStream"
           }
       ]
   }

5. 選擇下一步，然後選擇儲存變更。

設定應用程式

1. 在MyApplication頁面上，選擇設定。

2. 在應用程式程式碼位置區段中：

   • 對於 Amazon S3 儲存貯體，請選取先前為應用程式程式碼建立的儲存貯體。選擇「瀏覽」並選取正確的值區，然後選擇「選擇」。請勿選取值區名稱。

   • 對於 Amazon S3 物件的路徑，請輸入 managed-flink-pyflink-getting-started-1.0.0.zip。

3. 對於 [存取權限]，選擇 [建立/更新kinesis-analytics-MyApplication-us-east-1具有必要策略的IAM角色]

4. 移至「執行階段」屬性，並保留所有其他設定的預設值。

5. 選擇「新增項目」，然後新增下列每個參數：

   群組 ID	金鑰	值
   InputStream0	stream.name	ExampleInputStream
   InputStream0	flink.stream.initpos	LATEST
   InputStream0	aws.region	us-east-1
   OutputStream0	stream.name	ExampleOutputStream
   OutputStream0	aws.region	us-east-1
   kinesis.analytics.flink.run.options	python	main.py
   kinesis.analytics.flink.run.options	jarfile	lib/pyflink-dependencies.jar

6. 請勿修改任何其他區段，然後選擇 [儲存變更]。

執行應用程式

1. 在適用於 Apache Flink 的 Amazon 受管服務的主控台上，選擇我的應用程式，然後選擇執行。

2. 在下一頁的 [應用程式還原組態] 頁面上，選擇 [以最新快照執行]，然後選擇 [執行]。

   「應用程式中的狀態」 詳細資訊會從Ready應用程式啟動到再轉換為應用程式的Running時間。Starting

開啟 儀表板

1. 選擇「開啟阿帕奇 Flink 儀表板」。儀表板會在新頁面上開啟。

2. 在「執行工作」清單中，選擇您可以看到的單一工作。

   注意

   如果您設定執行階段屬性或編輯不正確的IAM原則，應用程式狀態可能會變成Running，但 Flink 儀表板會顯示工作正在持續重新啟動。如果應用程式設定錯誤或缺少存取外部資源的權限，這是常見的失敗案例。

   發生這種情況時，請檢查 Flink 儀表板中的「例外」標籤以查看問題的原因。

若要檢視測量結果

1. 在「重新整理」按鈕旁邊，從下拉式清單中選取 10 秒。

2. 當應用程式執行且健康狀態良好時，您可以看到正常執行時間指標不斷增加。

3. 完整重新啟動指標應為零。如果正在增加，則配置可能有問題。若要調查問題，請檢閱 Flink 儀表板上的「例外」標籤。

4. 在運作狀況良好的應用程式中，失敗的檢查點數量應為零。

   注意

   此儀表板會顯示一組固定的指標，精細度為 5 分鐘。您可以使用儀表板中的任何指標來建立自訂應用程式 CloudWatch 儀表板。

若要檢視輸出

1. 在 https://console.aws.amazon.com/Kinesis 處開啟室壁運動主控台。

2. 確認區域與您用來執行此自學課程的區域相同。依預設，它是美國東部-1 美國東部 (維吉尼亞北部)。如有必要，請變更「區域」。

3. 選擇「資料串流」。

4. 選取您要觀察的串流。在本教學課程中，使用 ExampleOutputStream。

5. 選擇「資料檢視器」標籤。

6. 選取任何 [碎片]，將 [最新] 保留為起始位置，然後選擇 [取得記錄]。您可能會看到「找不到此要求的記錄」錯誤訊息。如果是這樣，請選擇重試取得記錄。發佈至串流的最新記錄會顯示。

7. 在「資料」欄中選取值，以JSON格式檢查記錄的內容。

停止應用程式

1. 從「動作」下拉式清單中，選擇「停止」。

2. 「應用程式中的狀態」 詳細資訊Ready會從應Running用程式完全停止轉換為Stopping，然後再轉換為應用程式的

   注意

   別忘了也停止將資料從 Python 指令碼或 Kinesis 資料產生器傳送至輸入串流。