為 Apache 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. 導覽至 amazon-managed-service-for-apache-flink-examples/tree/main/java/GettingStarted 目錄。

觀察記錄

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

2. 確認區域與您執行此教學課程的位置相同，依預設為 us-east-1 美國東部 (維吉尼亞北部)。如果「區域」不相符，請變更該區域。

3. 選擇「資料串流」。

4. 選取您要觀察的串流，ExampleInputStream或 ExampleOutputStream.

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

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

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

上傳應用程式程式碼JAR檔

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

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

3. 選擇上傳。

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

5. 導覽至在上一個步驟中產生的JAR檔案：target/amazon-msf-java-stream-app-1.0.jar。

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

建立應用程式

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

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

3. 打開右側的菜單，然後選擇 Apache Flink 應用程序，然後創建流應用程序。或者，在初始頁面的 [開始使用] 容器中選擇 [建立串流應用程式]。

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

   • 選擇設定串流處理應用程式的方法：選擇「從頭開始建立」。

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

5. 設定應用程式

   • 應用程式名稱：輸入MyApplication。

   • 描述：輸入My java test app。

   • 存取應用程式資源：選擇 [建立/更新kinesis-analytics-MyApplication-us-east-1具有必要策略的IAM角色]。

6. 設定應用程式設定的範本

   • 範本：選擇 [開發]。

7. 選擇頁面底部的 [建立串流應用程式]。

若要編輯策略

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 物件的路徑，請輸入 amazon-msf-java-stream-app-1.0.jar。

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

4. 在 [執行階段屬性] 區段中，新增下列屬性。

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

   群組 ID	金鑰	值
   InputStream0	stream.name	ExampleInputStream
   InputStream0	aws.region	us-east-1
   OutputStream0	stream.name	ExampleOutputStream
   OutputStream0	aws.region	us-east-1

6. 請勿修改任何其他區段。

7. 選擇 Save changes (儲存變更)。

執行應用程式

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 資料產生器傳送至輸入串流。