Managed Service for Apache Flink 的早期版本資訊

1. 請確定您已安裝 Apache Maven，且JAVA_HOME環境變數指向 JDK而非 JRE。您可以使用以下命令來測試 Apache Maven 安裝：

   mvn -version

2. 下載 Apache Flink 版本 1.8.2 來源程式碼：

   wget https://archive.apache.org/dist/flink/flink-1.8.2/flink-1.8.2-src.tgz

3. 解壓縮 Apache Flink 來源程式碼：

   tar -xvf flink-1.8.2-src.tgz

4. 切換到 Apache Flink 來源程式碼目錄：

   cd flink-1.8.2

5. 編譯並安裝 Apache Flink：

   mvn clean install -Pinclude-kinesis -DskipTests

   注意

   如果您在 Microsoft 視窗中編譯 Flink，則需要添加 -Drat.skip=true 參數。

若要註冊 AWS 帳戶

1. 開啟https://portal.aws.amazon.com/billing/註冊 。

2. 請遵循線上指示進行。

   部分註冊程序需接收來電，並在電話鍵盤輸入驗證碼。

   當您註冊 時 AWS 帳戶，AWS 帳戶根使用者會建立 。根使用者有權存取該帳戶中的所有 AWS 服務 和資源。作為安全最佳實務，請將管理存取權指派給使用者，並且僅使用根使用者來執行需要根使用者存取權的任務。

保護您的 AWS 帳戶根使用者

1. 選擇根使用者並輸入 AWS 帳戶 您的電子郵件地址，以帳戶擁有者AWS Management Console身分登入 。在下一頁中，輸入您的密碼。

   如需使用根使用者登入的說明，請參閱 AWS 登入 使用者指南中的以根使用者身分登入。

2. 為您的根使用者開啟多重要素驗證 （MFA）。

   如需指示，請參閱 IAM 使用者指南 中的為您的 AWS 帳戶 根使用者 （主控台） 啟用虛擬MFA裝置。

建立具有管理存取權的使用者

1. 啟用IAM身分中心。

   如需指示，請參閱 AWS IAM Identity Center 使用者指南中的啟用 AWS IAM Identity Center。

2. 在 IAM Identity Center 中，將管理存取權授予使用者。

   如需使用 IAM Identity Center 目錄 作為身分來源的教學課程，請參閱 AWS IAM Identity Center 使用者指南 中的使用預設值設定使用者存取權 IAM Identity Center 目錄。

以具有管理存取權的使用者身分登入

• 若要使用 IAM Identity Center 使用者登入，請使用您建立 IAM Identity Center 使用者時URL傳送到您電子郵件地址的登入。

  如需使用 IAM Identity Center 使用者登入的協助，請參閱 AWS 登入 使用者指南 中的登入 AWS 存取入口網站。

指派存取權給其他使用者

1. 在 IAM Identity Center 中，建立遵循套用最低權限許可最佳實務的許可集。

   如需指示，請參閱《AWS IAM Identity Center 使用者指南》中的建立許可集。

2. 將使用者指派至群組，然後對該群組指派單一登入存取權。

   如需指示，請參閱《AWS IAM Identity Center 使用者指南》中的新增群組。

若要設定 AWS CLI

1. 下載和設定 AWS CLI。如需相關指示，請參閱《AWS Command Line Interface 使用者指南》中的下列主題：

   • 安裝 AWS Command Line Interface

   • 設定 AWS CLI

2. 在 AWS CLI config 檔案中為管理員使用者新增具名設定檔。當您執行 AWS CLI 命令時，使用此設定檔。如需具名描述檔的詳細資訊，請參閱《AWS Command Line Interface 使用者指南》中的具名描述檔。

   [profile adminuser]
   aws_access_key_id = adminuser access key ID
   aws_secret_access_key = adminuser secret access key
   region = aws-region

   如需可用 AWS 區域的清單，請參閱中的區域和端點Amazon Web Services 一般參考。

   注意

   本教學課程中的範例程式碼和命令使用美國西部 (奧勒岡) 區域。若要使用其他區域，請將本教學課程的程式碼和指令中的「區域」變更為您要使用的區域。

3. 在命令提示字元中輸入下列 help 命令，以驗證設定：

   aws help

建立資料串流 (AWS CLI)

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

   $ aws kinesis create-stream \
   --stream-name ExampleInputStream \
   --shard-count 1 \
   --region us-west-2 \
   --profile adminuser

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

   $ aws kinesis create-stream \
   --stream-name ExampleOutputStream \
   --shard-count 1 \
   --region us-west-2 \
   --profile adminuser

1. 使用下列內容建立名為 stock.py 的檔案：

   
       import datetime
       import json
       import random
       import boto3
       STREAM_NAME = "ExampleInputStream"
       def get_data():
           return {
               'event_time': datetime.datetime.now().isoformat(),
               'ticker': random.choice(['AAPL', 'AMZN', 'MSFT', 'INTC', 'TBV']),
               'price': round(random.random() * 100, 2)}
       def generate(stream_name, kinesis_client):
           while True:
               data = get_data()
               print(data)
               kinesis_client.put_record(
                   StreamName=stream_name,
                   Data=json.dumps(data),
                   PartitionKey="partitionkey")
       if __name__ == '__main__':
           generate(STREAM_NAME, boto3.client('kinesis', region_name='us-west-2')) 

2. 在教學課程後半段，您會執行 stock.py 指令碼來傳送資料至應用程式。

   $ python stock.py

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

   git clone https://github.com/aws-samples/amazon-kinesis-data-analytics-examples.git

2. 導覽至 amazon-kinesis-data-analytics-java-examples/GettingStarted 目錄。

編譯應用程式的程式碼

1. 若要使用應用程式程式碼，您可以將其編譯並封裝為 JAR 檔案。您可以使用下列兩種方式的其中之一，編譯和封裝您的程式碼：

   • 使用命令列 Maven 工具。在包含 JAR 檔案的目錄中執行下列命令來建立您的pom.xml檔案：

     mvn package -Dflink.version=1.13.2

   • 設定開發環境。如需詳細資訊，請參閱您的開發環境文件。

     注意

     提供的來源程式碼依賴於 Java 11 中的程式庫。

   您可以上傳套件作為JAR檔案，也可以壓縮套件並將其作為ZIP檔案上傳。如果您使用 建立應用程式 AWS CLI，您可以指定程式碼內容類型 （JAR 或 ZIP）。

2. 如果編譯時發生錯誤，請確認您的 JAVA_HOME 環境變數是否正確設定。

上傳應用程式的程式碼

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

2. 選擇建立儲存貯體。

3. 在儲存貯體名稱欄位中，輸入 ka-app-code-<username>。新增尾碼至儲存貯體名稱，例如您的使用者名稱，使其成為全域唯一的。選擇 Next (下一步)。

4. 在設定選項步驟中，保留原有設定並選擇 Next (下一步)。

5. 在設定許可步驟中，保留原有設定並選擇 Next (下一步)。

6. 選擇建立儲存貯體。

7. 在 Amazon S3 主控台中，選擇 ka-app-code-<username> 儲存貯體，然後選擇上傳 。

8. 在選取檔案步驟中，選擇 新增檔案。導覽至您在上一步驟中建立的 aws-kinesis-analytics-java-apps-1.0.jar 檔案。選擇 Next (下一步)。

9. 您不需要變更物件的任何設定，因此請選擇上傳。

1. 在 https://console.aws.amazon.com/flink 開啟 Managed Service for Apache Flink 主控台

2. 在 Managed Service for Apache Flink 儀表板上，選擇建立分析應用程式。

3. 在 Managed Service for Apache Flink - 建立應用程式頁面，提供應用程式詳細資訊，如下所示：

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

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

   • 對於​執行期，選擇 ​Apache Flink。

   • 將版本下拉式清單保留為 Apache Flink 1.13 版。

4. 針對存取許可 ，選擇建立/更新IAM角色 kinesis-analytics-MyApplication-us-west-2。

5. 選擇建立應用程式。

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

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

3. 在摘要頁面，選擇編輯政策。選擇 JSON 索引標籤。

4. 將下列政策範例的反白部分新增至政策。取代範例帳戶 IDs（012345678901） 搭配您的帳戶 ID。

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:s3:::ka-app-code-username/aws-kinesis-analytics-java-apps-1.0.jar"
               ]
           },
           {
               "Sid": "DescribeLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*"
               ]
           },
           {
               "Sid": "DescribeLogStreams",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogStreams"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:*"
               ]
           },
           {
               "Sid": "PutLogEvents",
               "Effect": "Allow",
               "Action": [
                   "logs:PutLogEvents"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:kinesis-analytics-log-stream"
               ]
           },
           {
               "Sid": "ReadInputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream"
           },
           {
               "Sid": "WriteOutputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleOutputStream"
           }
       ]
   }

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

2. 在設定應用程式頁面，提供程式碼位置：

   • 對於 Amazon S3 儲存貯體，請輸入 ka-app-code-<username>。

   • 對於 Amazon S3 物件的路徑，請輸入 aws-kinesis-analytics-java-apps-1.0.jar。

3. 在存取應用程式資源 下，針對存取許可 ，選擇建立/更新IAM角色 kinesis-analytics-MyApplication-us-west-2。

4. 輸入下列資料：

   群組 ID	金鑰	值
   ProducerConfigProperties	flink.inputstream.initpos	LATEST
   ProducerConfigProperties	aws.region	us-west-2
   ProducerConfigProperties	AggregationEnabled	false

5. 在監控下，確保監控指標層級設為應用程式。

6. 對於CloudWatch 記錄 ，選取啟用核取方塊。

7. 選擇更新。

建立 IAM 角色

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

2. 在導覽窗格中，選擇角色 、建立角色。

3. 在選取可信身分類型下，選擇 AWS 服務。在選擇將使用此角色的服務下，選擇 Kinesis。在 Select your use case (選取您的使用案例) 下，選擇 Kinesis Analytics (Kinesis 分析)。

   選擇下一步：許可。

4. 在連接許可政策頁面，選擇下一步：檢閱。您會在建立角色後連接許可政策。

5. 在建立角色頁面，輸入 MF-stream-rw-role 作為角色名稱。選擇建立角色。

   現在您已建立名為 的新IAM角色MF-stream-rw-role。您接著會更新角色的信任和許可政策。

6. 將 許可政策連接到角色。

   注意

   在此練習中，Managed Service for Apache Flink 擔任從 Kinesis 資料串流 (來源) 讀取資料並將輸出寫入另一個 Kinesis 資料串流的角色。因此您會連接在上一個步驟中建立的政策，建立許可政策。

   1. 在摘要頁面，選擇許可標籤。

   2. 選擇連接政策。

   3. 在搜尋方塊中，輸入 AKReadSourceStreamWriteSinkStream (您在上一節中建立的政策)。

   4. 選擇AKReadSourceStreamWriteSinkStream政策，然後選擇附加政策 。

1. 將下列JSON程式碼儲存至名為 的檔案create_request.json。將範例角色取代ARN為您先前建立的角色ARN的 。將儲存貯體尾碼 ARN （username） 取代為您在上一節中選擇的尾碼。使用您的帳戶 ID 取代服務執行角色中的範例帳戶 ID (012345678901)。

   {
       "ApplicationName": "test",
       "ApplicationDescription": "my java test app",
       "RuntimeEnvironment": "FLINK-1_15",
       "ServiceExecutionRole": "arn:aws:iam::012345678901:role/MF-stream-rw-role",
       "ApplicationConfiguration": {
           "ApplicationCodeConfiguration": {
               "CodeContent": {
                   "S3ContentLocation": {
                       "BucketARN": "arn:aws:s3:::ka-app-code-username",
                       "FileKey": "aws-kinesis-analytics-java-apps-1.0.jar"
                   }
               },
               "CodeContentType": "ZIPFILE"
           },
           "EnvironmentProperties":  { 
            "PropertyGroups": [ 
               { 
                  "PropertyGroupId": "ProducerConfigProperties",
                  "PropertyMap" : {
                       "flink.stream.initpos" : "LATEST",
                       "aws.region" : "us-west-2",
                       "AggregationEnabled" : "false"
                  }
               },
               { 
                  "PropertyGroupId": "ConsumerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2"
                  }
               }
            ]
         }
       }
   }
   
   

2. 使用前述請求執行 CreateApplication 動作以建立應用程式：

   aws kinesisanalyticsv2 create-application --cli-input-json file://create_request.json

啟動應用程式

1. 將下列JSON程式碼儲存至名為 的檔案start_request.json。

   {
       "ApplicationName": "test",
       "RunConfiguration": {
           "ApplicationRestoreConfiguration": { 
            "ApplicationRestoreType": "RESTORE_FROM_LATEST_SNAPSHOT"
            }
       }
   }
   

2. 以啟動應用程式的上述請求，執行 StartApplication 動作：

   aws kinesisanalyticsv2 start-application --cli-input-json file://start_request.json

停止應用程式

1. 將下列JSON程式碼儲存至名為 的檔案stop_request.json。

   {
      "ApplicationName": "test"
   }
   

2. 以停止應用程式的上述請求，執行 StopApplication 動作：

   aws kinesisanalyticsv2 stop-application --cli-input-json file://stop_request.json

更新應用程式的環境屬性

1. 將下列JSON程式碼儲存至名為 的檔案update_properties_request.json。

   {"ApplicationName": "test",
      "CurrentApplicationVersionId": 1,
      "ApplicationConfigurationUpdate": { 
         "EnvironmentPropertyUpdates": { 
            "PropertyGroups": [ 
               { 
                  "PropertyGroupId": "ProducerConfigProperties",
                  "PropertyMap" : {
                       "flink.stream.initpos" : "LATEST",
                       "aws.region" : "us-west-2",
                       "AggregationEnabled" : "false"
                  }
               },
               { 
                  "PropertyGroupId": "ConsumerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2"
                  }
               }
            ]
         }
      }
   }
   

2. 使用前述請求執行 UpdateApplication 動作以更新環境屬性：

   aws kinesisanalyticsv2 update-application --cli-input-json file://update_properties_request.json

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

2. 在 Managed Service for Apache Flink 面板中，選擇 MyApplication。

3. 在應用程式的頁面，選擇刪除，然後確認刪除。

1. 在 https://console.aws.amazon.com/flink 開啟 Managed Service for Apache Flink 主控台

2. 在 Kinesis Data Streams 面板中，選擇 ExampleInputStream。

3. 在 ExampleInputStream頁面中，選擇刪除 Kinesis Stream，然後確認刪除。

4. 在 Kinesis 串流頁面中，選擇 ExampleOutputStream，選擇動作 ，選擇刪除 ，然後確認刪除。

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

2. 選擇 ka-app-code-<username> 儲存貯體。

3. 選擇刪除，然後輸入儲存貯體名稱以確認刪除。

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

2. 在導覽列中，選擇政策。

3. 在篩選器控制項中，輸入 kinesis。

4. 選擇 kinesis-analytics-service-MyApplication-us-west-2 政策。

5. 選擇政策動作，然後選擇刪除。

6. 在導覽列中，選擇角色。

7. 選擇 kinesis-analytics-MyApplication-us-west-2 角色。

8. 選擇刪除角色，然後確認刪除。

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

2. 在導覽列中，選擇日誌。

3. 選擇 /aws/kinesis-analytics/MyApplication 日誌群組。

4. 選擇刪除日誌群組，然後確認刪除。

若要註冊 AWS 帳戶

1. 開啟https://portal.aws.amazon.com/billing/註冊 。

2. 請遵循線上指示進行。

   部分註冊程序需接收來電，並在電話鍵盤輸入驗證碼。

   當您註冊 時 AWS 帳戶，AWS 帳戶根使用者會建立 。根使用者有權存取該帳戶中的所有 AWS 服務 和資源。作為安全最佳實務，請將管理存取權指派給使用者，並且僅使用根使用者來執行需要根使用者存取權的任務。

保護您的 AWS 帳戶根使用者

1. 選擇根使用者並輸入 AWS 帳戶 您的電子郵件地址，以帳戶擁有者AWS Management Console身分登入 。在下一頁中，輸入您的密碼。

   如需使用根使用者登入的說明，請參閱 AWS 登入 使用者指南中的以根使用者身分登入。

2. 為您的根使用者開啟多重要素驗證 （MFA）。

   如需指示，請參閱 IAM 使用者指南 中的為您的 AWS 帳戶 根使用者 （主控台） 啟用虛擬MFA裝置。

建立具有管理存取權的使用者

1. 啟用IAM身分中心。

   如需指示，請參閱 AWS IAM Identity Center 使用者指南中的啟用 AWS IAM Identity Center。

2. 在 IAM Identity Center 中，將管理存取權授予使用者。

   如需使用 IAM Identity Center 目錄 做為身分來源的教學課程，請參閱 AWS IAM Identity Center 使用者指南 中的使用 設定使用者存取權 IAM Identity Center 目錄。

以具有管理存取權的使用者身分登入

• 若要使用 IAM Identity Center 使用者登入，請使用您建立 IAM Identity Center 使用者時URL傳送到您電子郵件地址的登入。

  如需使用 IAM Identity Center 使用者登入的協助，請參閱 AWS 登入 使用者指南 中的登入 AWS 存取入口網站。

指派存取權給其他使用者

1. 在 IAM Identity Center 中，建立遵循套用最低權限許可最佳實務的許可集。

   如需指示，請參閱《AWS IAM Identity Center 使用者指南》中的建立許可集。

2. 將使用者指派至群組，然後對該群組指派單一登入存取權。

   如需指示，請參閱《AWS IAM Identity Center 使用者指南》中的新增群組。

若要設定 AWS CLI

1. 下載和設定 AWS CLI。如需相關指示，請參閱《AWS Command Line Interface 使用者指南》中的下列主題：

   • 安裝 AWS Command Line Interface

   • 設定 AWS CLI

2. 在 AWS CLI config 檔案中為管理員使用者新增具名設定檔。當您執行 AWS CLI 命令時，使用此設定檔。如需具名描述檔的詳細資訊，請參閱《AWS Command Line Interface 使用者指南》中的具名描述檔。

   [profile adminuser]
   aws_access_key_id = adminuser access key ID
   aws_secret_access_key = adminuser secret access key
   region = aws-region

   如需可用 AWS 區域的清單，請參閱中的區域和端點Amazon Web Services 一般參考。

   注意

   本教學課程中的範例程式碼和命令使用美國西部 (奧勒岡) 區域。若要使用其他區域，請將本教學課程的程式碼和指令中的「區域」變更為您要使用的區域。

3. 在命令提示字元中輸入下列 help 命令，以驗證設定：

   aws help

建立資料串流 (AWS CLI)

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

   $ aws kinesis create-stream \
   --stream-name ExampleInputStream \
   --shard-count 1 \
   --region us-west-2 \
   --profile adminuser

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

   $ aws kinesis create-stream \
   --stream-name ExampleOutputStream \
   --shard-count 1 \
   --region us-west-2 \
   --profile adminuser

1. 使用下列內容建立名為 stock.py 的檔案：

    
   import datetime
   import json
   import random
   import boto3
   
   STREAM_NAME = "ExampleInputStream"
   
   
   def get_data():
       return {
           "EVENT_TIME": datetime.datetime.now().isoformat(),
           "TICKER": random.choice(["AAPL", "AMZN", "MSFT", "INTC", "TBV"]),
           "PRICE": round(random.random() * 100, 2),
       }
   
   
   def generate(stream_name, kinesis_client):
       while True:
           data = get_data()
           print(data)
           kinesis_client.put_record(
               StreamName=stream_name, Data=json.dumps(data), PartitionKey="partitionkey"
           )
   
   
   if __name__ == "__main__":
       generate(STREAM_NAME, boto3.client("kinesis"))
   

2. 在教學課程後半段，您會執行 stock.py 指令碼來傳送資料至應用程式。

   $ python stock.py

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

   git clone https://github.com/aws-samples/amazon-kinesis-data-analytics-examples.git

2. 導覽至 amazon-kinesis-data-analytics-java-examples/GettingStarted 目錄。

編譯應用程式的程式碼

1. 若要使用應用程式程式碼，您可以將其編譯並封裝為 JAR 檔案。您可以使用下列兩種方式的其中之一，編譯和封裝您的程式碼：

   • 使用命令列 Maven 工具。在包含 JAR 檔案的目錄中執行下列命令來建立您的pom.xml檔案：

     mvn package -Dflink.version=1.11.3

   • 設定開發環境。如需詳細資訊，請參閱您的開發環境文件。

     注意

     提供的來源程式碼依賴於 Java 11 中的程式庫。確保您專案的 Java 版本是 11。

   您可以上傳套件作為JAR檔案，也可以壓縮套件並將其作為ZIP檔案上傳。如果您使用 建立應用程式 AWS CLI，您可以指定程式碼內容類型 （JAR 或 ZIP）。

2. 如果編譯時發生錯誤，請確認您的 JAVA_HOME 環境變數是否正確設定。

上傳應用程式的程式碼

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

2. 選擇建立儲存貯體。

3. 在儲存貯體名稱欄位中，輸入 ka-app-code-<username>。新增尾碼至儲存貯體名稱，例如您的使用者名稱，使其成為全域唯一的。選擇 Next (下一步)。

4. 在設定選項步驟中，保留原有設定並選擇 Next (下一步)。

5. 在設定許可步驟中，保留原有設定並選擇 Next (下一步)。

6. 選擇建立儲存貯體。

7. 在 Amazon S3 主控台中，選擇 ka-app-code-<username> 儲存貯體，然後選擇上傳 。

8. 在選取檔案步驟中，選擇 新增檔案。導覽至您在上一步驟中建立的 aws-kinesis-analytics-java-apps-1.0.jar 檔案。選擇 Next (下一步)。

9. 您不需要變更物件的任何設定，因此請選擇上傳。

1. 在 https://console.aws.amazon.com/flink 開啟 Managed Service for Apache Flink 主控台

2. 在 Managed Service for Apache Flink 儀表板上，選擇建立分析應用程式。

3. 在 Managed Service for Apache Flink - 建立應用程式頁面，提供應用程式詳細資訊，如下所示：

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

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

   • 對於​執行期，選擇 ​Apache Flink。

   • 將版本下拉式清單保留為 Apache Flink 1.11 版 (建議版本)。

4. 針對存取許可 ，選擇建立/更新IAM角色 kinesis-analytics-MyApplication-us-west-2。

5. 選擇建立應用程式。

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

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

3. 在摘要頁面，選擇編輯政策。選擇 JSON 索引標籤。

4. 將下列政策範例的反白部分新增至政策。取代範例帳戶 IDs（012345678901） 搭配您的帳戶 ID。

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:s3:::ka-app-code-username/aws-kinesis-analytics-java-apps-1.0.jar"
               ]
           },
           {
               "Sid": "DescribeLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*"
               ]
           },
           {
               "Sid": "DescribeLogStreams",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogStreams"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:*"
               ]
           },
           {
               "Sid": "PutLogEvents",
               "Effect": "Allow",
               "Action": [
                   "logs:PutLogEvents"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:kinesis-analytics-log-stream"
               ]
           },
           {
               "Sid": "ReadInputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream"
           },
           {
               "Sid": "WriteOutputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleOutputStream"
           }
       ]
   }

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

2. 在設定應用程式頁面，提供程式碼位置：

   • 對於 Amazon S3 儲存貯體，請輸入 ka-app-code-<username>。

   • 對於 Amazon S3 物件的路徑，請輸入 aws-kinesis-analytics-java-apps-1.0.jar。

3. 在存取應用程式資源 下，針對存取許可 ，選擇建立/更新IAM角色 kinesis-analytics-MyApplication-us-west-2。

4. 在屬性下，為群組 ID輸入 ProducerConfigProperties。

5. 輸入以下應用程式屬性和數值：

   群組 ID	金鑰	值
   ProducerConfigProperties	flink.inputstream.initpos	LATEST
   ProducerConfigProperties	aws.region	us-west-2
   ProducerConfigProperties	AggregationEnabled	false

6. 在監控下，確保監控指標層級設為應用程式。

7. 對於CloudWatch 記錄 ，選取啟用核取方塊。

8. 選擇更新。

建立 IAM 角色

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

2. 在導覽窗格中，選擇角色 、建立角色。

3. 在選取可信身分類型下，選擇 AWS 服務。在選擇將使用此角色的服務下，選擇 Kinesis。在 Select your use case (選取您的使用案例) 下，選擇 Kinesis Analytics (Kinesis 分析)。

   選擇下一步：許可。

4. 在連接許可政策頁面，選擇下一步：檢閱。您會在建立角色後連接許可政策。

5. 在建立角色頁面，輸入 MF-stream-rw-role 作為角色名稱。選擇建立角色。

   現在您已建立名為 的新IAM角色MF-stream-rw-role。您接著會更新角色的信任和許可政策。

6. 將 許可政策連接到角色。

   注意

   在此練習中，Managed Service for Apache Flink 擔任從 Kinesis 資料串流 (來源) 讀取資料並將輸出寫入另一個 Kinesis 資料串流的角色。因此您會連接在上一個步驟中建立的政策，建立許可政策。

   1. 在摘要頁面，選擇許可標籤。

   2. 選擇連接政策。

   3. 在搜尋方塊中，輸入 AKReadSourceStreamWriteSinkStream (您在上一節中建立的政策)。

   4. 選擇AKReadSourceStreamWriteSinkStream政策，然後選擇附加政策 。

1. 將下列JSON程式碼儲存至名為 的檔案create_request.json。將範例角色取代ARN為您先前建立的角色ARN的 。將儲存貯體尾碼 ARN （username） 取代為您在上一節中選擇的尾碼。使用您的帳戶 ID 取代服務執行角色中的範例帳戶 ID (012345678901)。

   {
       "ApplicationName": "test",
       "ApplicationDescription": "my java test app",
       "RuntimeEnvironment": "FLINK-1_11",
       "ServiceExecutionRole": "arn:aws:iam::012345678901:role/MF-stream-rw-role",
       "ApplicationConfiguration": {
           "ApplicationCodeConfiguration": {
               "CodeContent": {
                   "S3ContentLocation": {
                       "BucketARN": "arn:aws:s3:::ka-app-code-username",
                       "FileKey": "aws-kinesis-analytics-java-apps-1.0.jar"
                   }
               },
               "CodeContentType": "ZIPFILE"
           },
           "EnvironmentProperties":  { 
            "PropertyGroups": [ 
               { 
                  "PropertyGroupId": "ProducerConfigProperties",
                  "PropertyMap" : {
                       "flink.stream.initpos" : "LATEST",
                       "aws.region" : "us-west-2",
                       "AggregationEnabled" : "false"
                  }
               },
               { 
                  "PropertyGroupId": "ConsumerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2"
                  }
               }
            ]
         }
       }
   }
   
   

2. 使用前述請求執行 CreateApplication 動作以建立應用程式：

   aws kinesisanalyticsv2 create-application --cli-input-json file://create_request.json

啟動應用程式

1. 將下列JSON程式碼儲存至名為 的檔案start_request.json。

   {
       "ApplicationName": "test",
       "RunConfiguration": {
           "ApplicationRestoreConfiguration": { 
            "ApplicationRestoreType": "RESTORE_FROM_LATEST_SNAPSHOT"
            }
       }
   }
   

2. 以啟動應用程式的上述請求，執行 StartApplication 動作：

   aws kinesisanalyticsv2 start-application --cli-input-json file://start_request.json

停止應用程式

1. 將下列JSON程式碼儲存至名為 的檔案stop_request.json。

   
   {
       "ApplicationName": "test"
   }
   

2. 以停止應用程式的上述請求，執行 StopApplication 動作：

   aws kinesisanalyticsv2 stop-application --cli-input-json file://stop_request.json

更新應用程式的環境屬性

1. 將下列JSON程式碼儲存至名為 的檔案update_properties_request.json。

   {"ApplicationName": "test",
      "CurrentApplicationVersionId": 1,
      "ApplicationConfigurationUpdate": { 
         "EnvironmentPropertyUpdates": { 
            "PropertyGroups": [ 
               { 
                  "PropertyGroupId": "ProducerConfigProperties",
                  "PropertyMap" : {
                       "flink.stream.initpos" : "LATEST",
                       "aws.region" : "us-west-2",
                       "AggregationEnabled" : "false"
                  }
               },
               { 
                  "PropertyGroupId": "ConsumerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2"
                  }
               }
            ]
         }
      }
   }
   

2. 使用前述請求執行 UpdateApplication 動作以更新環境屬性：

   aws kinesisanalyticsv2 update-application --cli-input-json file://update_properties_request.json

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

2. 在 Managed Service for Apache Flink 面板中，選擇 MyApplication。

3. 在應用程式的頁面，選擇刪除，然後確認刪除。

1. 在 https://console.aws.amazon.com/flink 開啟 Managed Service for Apache Flink 主控台

2. 在 Kinesis Data Streams 面板中，選擇 ExampleInputStream。

3. 在 ExampleInputStream頁面中，選擇刪除 Kinesis Stream，然後確認刪除。

4. 在 Kinesis 串流頁面中，選擇 ExampleOutputStream，選擇動作 ，選擇刪除 ，然後確認刪除。

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

2. 選擇 ka-app-code-<username> 儲存貯體。

3. 選擇刪除，然後輸入儲存貯體名稱以確認刪除。

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

2. 在導覽列中，選擇政策。

3. 在篩選器控制項中，輸入 kinesis。

4. 選擇 kinesis-analytics-service-MyApplication-us-west-2 政策。

5. 選擇政策動作，然後選擇刪除。

6. 在導覽列中，選擇角色。

7. 選擇 kinesis-analytics-MyApplication-us-west-2 角色。

8. 選擇刪除角色，然後確認刪除。

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

2. 在導覽列中，選擇日誌。

3. 選擇 /aws/kinesis-analytics/MyApplication 日誌群組。

4. 選擇刪除日誌群組，然後確認刪除。

若要註冊 AWS 帳戶

1. 開啟https://portal.aws.amazon.com/billing/註冊 。

2. 請遵循線上指示進行。

   部分註冊程序需接收來電，並在電話鍵盤輸入驗證碼。

   當您註冊 時 AWS 帳戶，AWS 帳戶根使用者會建立 。根使用者有權存取該帳戶中的所有 AWS 服務 和資源。作為安全最佳實務，請將管理存取權指派給使用者，並且僅使用根使用者來執行需要根使用者存取權的任務。

保護您的 AWS 帳戶根使用者

1. 選擇根使用者並輸入 AWS 帳戶 您的電子郵件地址，以帳戶擁有者AWS Management Console身分登入 。在下一頁中，輸入您的密碼。

   如需使用根使用者登入的說明，請參閱 AWS 登入 使用者指南中的以根使用者身分登入。

2. 為您的根使用者開啟多重要素驗證 （MFA）。

   如需指示，請參閱 IAM 使用者指南 中的為 AWS 帳戶 根使用者 （主控台） 啟用虛擬MFA裝置。

建立具有管理存取權的使用者

1. 啟用IAM身分中心。

   如需指示，請參閱 AWS IAM Identity Center 使用者指南中的啟用 AWS IAM Identity Center。

2. 在 IAM Identity Center 中，將管理存取權授予使用者。

   如需使用 IAM Identity Center 目錄 作為身分來源的教學課程，請參閱 AWS IAM Identity Center 使用者指南 中的使用 設定使用者存取權 IAM Identity Center 目錄。

以具有管理存取權的使用者身分登入

• 若要使用 IAM Identity Center 使用者登入，請使用您建立 IAM Identity Center 使用者時URL傳送到您電子郵件地址的登入。

  如需使用 IAM Identity Center 使用者登入的協助，請參閱 AWS 登入 使用者指南 中的登入 AWS 存取入口網站。

指派存取權給其他使用者

1. 在 IAM Identity Center 中，建立遵循套用最低權限許可最佳實務的許可集。

   如需指示，請參閱《AWS IAM Identity Center 使用者指南》中的建立許可集。

2. 將使用者指派至群組，然後對該群組指派單一登入存取權。

   如需指示，請參閱《AWS IAM Identity Center 使用者指南》中的新增群組。

若要設定 AWS CLI

1. 下載和設定 AWS CLI。如需相關指示，請參閱《AWS Command Line Interface 使用者指南》中的下列主題：

   • 安裝 AWS Command Line Interface

   • 設定 AWS CLI

2. 在 AWS CLI config 檔案中為管理員使用者新增具名設定檔。當您執行 AWS CLI 命令時，使用此設定檔。如需具名描述檔的詳細資訊，請參閱《AWS Command Line Interface 使用者指南》中的具名描述檔。

   [profile adminuser]
   aws_access_key_id = adminuser access key ID
   aws_secret_access_key = adminuser secret access key
   region = aws-region

   如需可用區域的清單，請參閱 Amazon Web Services 一般參考 中的區域與端點。

   注意

   本教學課程中的範例程式碼和命令使用美國西部 (奧勒岡) 區域。若要使用不同的 AWS 區域，請將本教學課程程式碼和命令中的 區域變更為您要使用的 區域。

3. 在命令提示字元中輸入下列 help 命令，以驗證設定：

   aws help

建立資料串流 (AWS CLI)

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

   $ aws kinesis create-stream \
   --stream-name ExampleInputStream \
   --shard-count 1 \
   --region us-west-2 \
   --profile adminuser

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

   $ aws kinesis create-stream \
   --stream-name ExampleOutputStream \
   --shard-count 1 \
   --region us-west-2 \
   --profile adminuser

1. 使用下列內容建立名為 stock.py 的檔案：

    
   import datetime
   import json
   import random
   import boto3
   
   STREAM_NAME = "ExampleInputStream"
   
   
   def get_data():
       return {
           "EVENT_TIME": datetime.datetime.now().isoformat(),
           "TICKER": random.choice(["AAPL", "AMZN", "MSFT", "INTC", "TBV"]),
           "PRICE": round(random.random() * 100, 2),
       }
   
   
   def generate(stream_name, kinesis_client):
       while True:
           data = get_data()
           print(data)
           kinesis_client.put_record(
               StreamName=stream_name, Data=json.dumps(data), PartitionKey="partitionkey"
           )
   
   
   if __name__ == "__main__":
       generate(STREAM_NAME, boto3.client("kinesis"))
   

2. 在教學課程後半段，您會執行 stock.py 指令碼來傳送資料至應用程式。

   $ python stock.py

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

   git clone https://github.com/aws-samples/amazon-kinesis-data-analytics-examples.git

2. 導覽至 amazon-kinesis-data-analytics-java-examples/GettingStarted_1_8 目錄。

編譯應用程式的程式碼

1. 若要使用應用程式程式碼，您可以編譯應用程式程式碼並將其封裝為 JAR 檔案。您可以使用下列兩種方式的其中之一，編譯和封裝您的程式碼：

   • 使用命令列 Maven 工具。在包含 JAR 檔案的目錄中執行下列命令來建立您的pom.xml檔案：

     mvn package -Dflink.version=1.8.2

   • 設定開發環境。如需詳細資訊，請參閱您的開發環境文件。

     注意

     提供的來源程式碼依賴於 Java 1.8 中的程式庫。確保您專案的 Java 版本是 1.8。

   您可以上傳套件作為JAR檔案，也可以壓縮套件並將其作為ZIP檔案上傳。如果您使用 建立應用程式 AWS CLI，您可以指定程式碼內容類型 （JAR 或 ZIP）。

2. 如果編譯時發生錯誤，請確認您的 JAVA_HOME 環境變數是否正確設定。

上傳應用程式的程式碼

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

2. 選擇建立儲存貯體。

3. 在儲存貯體名稱欄位中，輸入 ka-app-code-<username>。新增尾碼至儲存貯體名稱，例如您的使用者名稱，使其成為全域唯一的。選擇 Next (下一步)。

4. 在設定選項步驟中，保留原有設定並選擇 Next (下一步)。

5. 在設定許可步驟中，保留原有設定並選擇 Next (下一步)。

6. 選擇建立儲存貯體。

7. 在 Amazon S3 主控台中，選擇 ka-app-code-<username> 儲存貯體，然後選擇上傳 。

8. 在選取檔案步驟中，選擇 新增檔案。導覽至您在上一步驟中建立的 aws-kinesis-analytics-java-apps-1.0.jar 檔案。選擇 Next (下一步)。

9. 您不需要變更物件的任何設定，因此請選擇上傳。

1. 在 https://console.aws.amazon.com/flink 開啟 Managed Service for Apache Flink 主控台

2. 在 Managed Service for Apache Flink 儀表板上，選擇建立分析應用程式。

3. 在 Managed Service for Apache Flink - 建立應用程式頁面，提供應用程式詳細資訊，如下所示：

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

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

   • 對於​執行期，選擇 ​Apache Flink。

   • 將版本下拉式清單保留為 Apache Flink 1.8 版 (建議版本)。

4. 針對存取許可 ，選擇建立/更新IAM角色 kinesis-analytics-MyApplication-us-west-2。

5. 選擇建立應用程式。

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

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

3. 在摘要頁面，選擇編輯政策。選擇 JSON 索引標籤。

4. 將下列政策範例的反白部分新增至政策。取代範例帳戶 IDs（012345678901） 搭配您的帳戶 ID。

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:s3:::ka-app-code-username/aws-kinesis-analytics-java-apps-1.0.jar"
               ]
           },
           {
               "Sid": "DescribeLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*"
               ]
           },
           {
               "Sid": "DescribeLogStreams",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogStreams"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:*"
               ]
           },
           {
               "Sid": "PutLogEvents",
               "Effect": "Allow",
               "Action": [
                   "logs:PutLogEvents"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:kinesis-analytics-log-stream"
               ]
           },
           {
               "Sid": "ReadInputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream"
           },
           {
               "Sid": "WriteOutputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleOutputStream"
           }
       ]
   }

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

2. 在設定應用程式頁面，提供程式碼位置：

   • 對於 Amazon S3 儲存貯體，請輸入 ka-app-code-<username>。

   • 對於 Amazon S3 物件的路徑，請輸入 aws-kinesis-analytics-java-apps-1.0.jar。

3. 在存取應用程式資源 下，針對存取許可 ，選擇建立/更新IAM角色 kinesis-analytics-MyApplication-us-west-2。

4. 輸入以下應用程式屬性和數值：

   群組 ID	金鑰	值
   ProducerConfigProperties	flink.inputstream.initpos	LATEST
   ProducerConfigProperties	aws.region	us-west-2
   ProducerConfigProperties	AggregationEnabled	false

5. 在監控下，確保監控指標層級設為應用程式。

6. 對於CloudWatch 記錄 ，選取啟用核取方塊。

7. 選擇更新。

1. 在頁面上MyApplication，選擇執行 。確認動作。

2. 應用程式執行時，重新整理頁面。主控台會顯示 Application graph (應用程式圖形)。

建立 IAM 角色

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

2. 在導覽窗格中，選擇角色 、建立角色。

3. 在選取可信身分類型下，選擇 AWS 服務。在選擇將使用此角色的服務下，選擇 Kinesis。在 Select your use case (選取您的使用案例) 下，選擇 Kinesis Analytics (Kinesis 分析)。

   選擇下一步：許可。

4. 在連接許可政策頁面，選擇下一步：檢閱。您會在建立角色後連接許可政策。

5. 在建立角色頁面，輸入 MF-stream-rw-role 作為角色名稱。選擇建立角色。

   現在您已建立名為 的新IAM角色MF-stream-rw-role。您接著會更新角色的信任和許可政策。

6. 將 許可政策連接到角色。

   注意

   在此練習中，Managed Service for Apache Flink 擔任從 Kinesis 資料串流 (來源) 讀取資料並將輸出寫入另一個 Kinesis 資料串流的角色。因此您會連接在上一個步驟中建立的政策，建立許可政策。

   1. 在摘要頁面，選擇許可標籤。

   2. 選擇連接政策。

   3. 在搜尋方塊中，輸入 AKReadSourceStreamWriteSinkStream (您在上一節中建立的政策)。

   4. 選擇AKReadSourceStreamWriteSinkStream政策，然後選擇附加政策 。

1. 將下列JSON程式碼儲存至名為 的檔案create_request.json。將範例角色取代ARN為您先前建立的角色ARN的 。將儲存貯體尾碼 ARN （username） 取代為您在上一節中選擇的尾碼。使用您的帳戶 ID 取代服務執行角色中的範例帳戶 ID (012345678901)。

   {
       "ApplicationName": "test",
       "ApplicationDescription": "my java test app",
       "RuntimeEnvironment": "FLINK-1_8",
       "ServiceExecutionRole": "arn:aws:iam::012345678901:role/MF-stream-rw-role",
       "ApplicationConfiguration": {
           "ApplicationCodeConfiguration": {
               "CodeContent": {
                   "S3ContentLocation": {
                       "BucketARN": "arn:aws:s3:::ka-app-code-username",
                       "FileKey": "aws-kinesis-analytics-java-apps-1.0.jar"
                   }
               },
               "CodeContentType": "ZIPFILE"
           },
           "EnvironmentProperties":  { 
            "PropertyGroups": [ 
               { 
                  "PropertyGroupId": "ProducerConfigProperties",
                  "PropertyMap" : {
                       "flink.stream.initpos" : "LATEST",
                       "aws.region" : "us-west-2",
                       "AggregationEnabled" : "false"
                  }
               },
               { 
                  "PropertyGroupId": "ConsumerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2"
                  }
               }
            ]
         }
       }
   }
   
   

2. 使用前述請求執行 CreateApplication 動作以建立應用程式：

   aws kinesisanalyticsv2 create-application --cli-input-json file://create_request.json

啟動應用程式

1. 將下列JSON程式碼儲存至名為 的檔案start_request.json。

   {
       "ApplicationName": "test",
       "RunConfiguration": {
           "ApplicationRestoreConfiguration": { 
            "ApplicationRestoreType": "RESTORE_FROM_LATEST_SNAPSHOT"
            }
       }
   }
   

2. 以啟動應用程式的上述請求，執行 StartApplication 動作：

   aws kinesisanalyticsv2 start-application --cli-input-json file://start_request.json

停止應用程式

1. 將下列JSON程式碼儲存至名為 的檔案stop_request.json。

   
   {
      "ApplicationName": "test"
   }
   

2. 以停止應用程式的上述請求，執行 StopApplication 動作：

   aws kinesisanalyticsv2 stop-application --cli-input-json file://stop_request.json

更新應用程式的環境屬性

1. 將下列JSON程式碼儲存至名為 的檔案update_properties_request.json。

   {"ApplicationName": "test",
      "CurrentApplicationVersionId": 1,
      "ApplicationConfigurationUpdate": { 
         "EnvironmentPropertyUpdates": { 
            "PropertyGroups": [ 
               { 
                  "PropertyGroupId": "ProducerConfigProperties",
                  "PropertyMap" : {
                       "flink.stream.initpos" : "LATEST",
                       "aws.region" : "us-west-2",
                       "AggregationEnabled" : "false"
                  }
               },
               { 
                  "PropertyGroupId": "ConsumerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2"
                  }
               }
            ]
         }
      }
   }
   

2. 使用前述請求執行 UpdateApplication 動作以更新環境屬性：

   aws kinesisanalyticsv2 update-application --cli-input-json file://update_properties_request.json

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

2. 在 Managed Service for Apache Flink 面板中，選擇 MyApplication。

3. 選擇設定。

4. 在快照區段中，選擇停用，然後選擇更新。

5. 在應用程式的頁面，選擇刪除，然後確認刪除。

1. 在 https://console.aws.amazon.com/flink 開啟 Managed Service for Apache Flink 主控台

2. 在 Kinesis Data Streams 面板中，選擇 ExampleInputStream。

3. 在ExampleInputStream頁面中，選擇刪除 Kinesis Stream，然後確認刪除。

4. 在 Kinesis 串流頁面中，選擇 ExampleOutputStream，選擇動作 ，選擇刪除 ，然後確認刪除。

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

2. 選擇 ka-app-code-<username> 儲存貯體。

3. 選擇刪除，然後輸入儲存貯體名稱以確認刪除。

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

2. 在導覽列中，選擇政策。

3. 在篩選器控制項中，輸入 kinesis。

4. 選擇 kinesis-analytics-service-MyApplication-us-west-2 政策。

5. 選擇政策動作，然後選擇刪除。

6. 在導覽列中，選擇角色。

7. 選擇 kinesis-analytics-MyApplication-us-west-2 角色。

8. 選擇刪除角色，然後確認刪除。

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

2. 在導覽列中，選擇日誌。

3. 選擇 /aws/kinesis-analytics/MyApplication 日誌群組。

4. 選擇刪除日誌群組，然後確認刪除。

若要註冊 AWS 帳戶

1. 開啟https://portal.aws.amazon.com/billing/註冊 。

2. 請遵循線上指示進行。

   部分註冊程序需接收來電，並在電話鍵盤輸入驗證碼。

   當您註冊 時 AWS 帳戶，AWS 帳戶根使用者會建立 。根使用者有權存取該帳戶中的所有 AWS 服務 和資源。作為安全最佳實務，請將管理存取權指派給使用者，並且僅使用根使用者來執行需要根使用者存取權的任務。

保護您的 AWS 帳戶根使用者

1. 選擇根使用者並輸入 AWS 帳戶 您的電子郵件地址，以帳戶擁有者AWS Management Console身分登入 。在下一頁中，輸入您的密碼。

   如需使用根使用者登入的說明，請參閱 AWS 登入 使用者指南中的以根使用者身分登入。

2. 為您的根使用者開啟多重要素驗證 （MFA）。

   如需指示，請參閱 IAM 使用者指南 中的為 AWS 帳戶 根使用者 （主控台） 啟用虛擬MFA裝置。

建立具有管理存取權的使用者

1. 啟用IAM身分中心。

   如需指示，請參閱 AWS IAM Identity Center 使用者指南中的啟用 AWS IAM Identity Center。

2. 在 IAM Identity Center 中，將管理存取權授予使用者。

   如需使用 IAM Identity Center 目錄 作為身分來源的教學課程，請參閱 AWS IAM Identity Center 使用者指南 中的使用 設定使用者存取權 IAM Identity Center 目錄。

以具有管理存取權的使用者身分登入

• 若要使用 IAM Identity Center 使用者登入，請使用您建立 IAM Identity Center 使用者時URL傳送到您電子郵件地址的登入。

  如需使用 IAM Identity Center 使用者登入的協助，請參閱 AWS 登入 使用者指南 中的登入 AWS 存取入口網站。

指派存取權給其他使用者

1. 在 IAM Identity Center 中，建立遵循套用最低權限許可最佳實務的許可集。

   如需指示，請參閱《AWS IAM Identity Center 使用者指南》中的建立許可集。

2. 將使用者指派至群組，然後對該群組指派單一登入存取權。

   如需指示，請參閱《AWS IAM Identity Center 使用者指南》中的新增群組。

若要設定 AWS CLI

1. 下載和設定 AWS CLI。如需相關指示，請參閱《AWS Command Line Interface 使用者指南》中的下列主題：

   • 安裝 AWS Command Line Interface

   • 設定 AWS CLI

2. 在 AWS CLI config 檔案中為管理員使用者新增具名設定檔。當您執行 AWS CLI 命令時，使用此設定檔。如需具名描述檔的詳細資訊，請參閱《AWS Command Line Interface 使用者指南》中的具名描述檔。

   [profile adminuser]
   aws_access_key_id = adminuser access key ID
   aws_secret_access_key = adminuser secret access key
   region = aws-region

   如需可用 AWS 區域的清單，請參閱中的區域和端點Amazon Web Services 一般參考。

   注意

   本教學課程中的範例程式碼和命令使用美國西部 (奧勒岡) 區域。若要使用其他區域，請將本教學課程的程式碼和指令中的「區域」變更為您要使用的區域。

3. 在命令提示字元中輸入下列 help 命令，以驗證設定：

   aws help

建立資料串流 (AWS CLI)

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

   $ aws kinesis create-stream \
   --stream-name ExampleInputStream \
   --shard-count 1 \
   --region us-west-2 \
   --profile adminuser

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

   $ aws kinesis create-stream \
   --stream-name ExampleOutputStream \
   --shard-count 1 \
   --region us-west-2 \
   --profile adminuser

1. 使用下列內容建立名為 stock.py 的檔案：

    
   import datetime
   import json
   import random
   import boto3
   
   STREAM_NAME = "ExampleInputStream"
   
   
   def get_data():
       return {
           "EVENT_TIME": datetime.datetime.now().isoformat(),
           "TICKER": random.choice(["AAPL", "AMZN", "MSFT", "INTC", "TBV"]),
           "PRICE": round(random.random() * 100, 2),
       }
   
   
   def generate(stream_name, kinesis_client):
       while True:
           data = get_data()
           print(data)
           kinesis_client.put_record(
               StreamName=stream_name, Data=json.dumps(data), PartitionKey="partitionkey"
           )
   
   
   if __name__ == "__main__":
       generate(STREAM_NAME, boto3.client("kinesis"))
   

2. 在教學課程後半段，您會執行 stock.py 指令碼來傳送資料至應用程式。

   $ python stock.py

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

   git clone https://github.com/aws-samples/amazon-kinesis-data-analytics-examples.git

2. 導覽至 amazon-kinesis-data-analytics-java-examples/GettingStarted_1_6 目錄。

編譯應用程式的程式碼

1. 若要使用應用程式程式碼，您可以編譯應用程式程式碼並將其封裝為 JAR 檔案。您可以使用下列兩種方式的其中之一，編譯和封裝您的程式碼：

   • 使用命令列 Maven 工具。在包含 JAR 檔案的目錄中執行下列命令來建立您的pom.xml檔案：

     mvn package

     注意

     Managed Service for Apache Flink Runtime 1.0.1 版本不需要 -Dflink.version 參數；只有版本 1.1.0 及更新版本才需要此參數。如需詳細資訊，請參閱指定應用程式的 Apache Flink 版本。

   • 設定開發環境。如需詳細資訊，請參閱您的開發環境文件。

   您可以上傳套件作為JAR檔案，也可以壓縮套件並將其作為ZIP檔案上傳。如果您使用 建立應用程式 AWS CLI，您可以指定程式碼內容類型 （JAR 或 ZIP）。

2. 如果編譯時發生錯誤，請確認您的 JAVA_HOME 環境變數是否正確設定。

上傳應用程式的程式碼

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

2. 選擇建立儲存貯體。

3. 在儲存貯體名稱欄位中，輸入 ka-app-code-<username>。新增尾碼至儲存貯體名稱，例如您的使用者名稱，使其成為全域唯一的。選擇 Next (下一步)。

4. 在設定選項步驟中，保留原有設定並選擇 Next (下一步)。

5. 在設定許可步驟中，保留原有設定並選擇 Next (下一步)。

6. 選擇建立儲存貯體。

7. 在 Amazon S3 主控台中，選擇 ka-app-code-<username> 儲存貯體，然後選擇上傳 。

8. 在選取檔案步驟中，選擇 新增檔案。導覽至您在上一步驟中建立的 aws-kinesis-analytics-java-apps-1.0.jar 檔案。選擇 Next (下一步)。

9. 在設定許可步驟中，保留原有設定。選擇 Next (下一步)。

10. 在設定屬性步驟中，保留原有設定。選擇上傳。

1. 在 https://console.aws.amazon.com/flink 開啟 Managed Service for Apache Flink 主控台

2. 在 Managed Service for Apache Flink 儀表板上，選擇建立分析應用程式。

3. 在 Managed Service for Apache Flink - 建立應用程式頁面，提供應用程式詳細資訊，如下所示：

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

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

   • 對於​執行期，選擇 ​Apache Flink。

     注意

     Managed Service for Apache Flink 使用 Apache Flink 1.8.2 或 1.6.2 版。

   • 將版本下拉式清單變更為 Apache Flink 1.6。

4. 針對存取許可 ，選擇建立/更新IAM角色 kinesis-analytics-MyApplication-us-west-2。

5. 選擇建立應用程式。

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

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

3. 在摘要頁面，選擇編輯政策。選擇 JSON 索引標籤。

4. 將下列政策範例的反白部分新增至政策。取代範例帳戶 IDs（012345678901） 搭配您的帳戶 ID。

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:s3:::ka-app-code-username/java-getting-started-1.0.jar"
               ]
           },
           {
               "Sid": "DescribeLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*"
               ]
           },
           {
               "Sid": "DescribeLogStreams",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogStreams"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:*"
               ]
           },
           {
               "Sid": "PutLogEvents",
               "Effect": "Allow",
               "Action": [
                   "logs:PutLogEvents"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:kinesis-analytics-log-stream"
               ]
           },
           {
               "Sid": "ReadInputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream"
           },
           {
               "Sid": "WriteOutputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleOutputStream"
           }
       ]
   }

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

2. 在設定應用程式頁面，提供程式碼位置：

   • 對於 Amazon S3 儲存貯體，請輸入 ka-app-code-<username>。

   • 對於 Amazon S3 物件的路徑，請輸入 java-getting-started-1.0.jar。

3. 在存取應用程式資源 下，針對存取許可 ，選擇建立/更新IAM角色 kinesis-analytics-MyApplication-us-west-2。

4. 輸入以下應用程式屬性和數值：

   群組 ID	金鑰	值
   ProducerConfigProperties	flink.inputstream.initpos	LATEST
   ProducerConfigProperties	aws.region	us-west-2
   ProducerConfigProperties	AggregationEnabled	false

5. 在監控下，確保監控指標層級設為應用程式。

6. 對於CloudWatch 記錄 ，選取啟用核取方塊。

7. 選擇更新。

1. 在頁面上MyApplication，選擇執行 。確認動作。

2. 應用程式執行時，重新整理頁面。主控台會顯示 Application graph (應用程式圖形)。

建立 IAM 角色

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

2. 在導覽窗格中，選擇角色 、建立角色。

3. 在選取可信身分類型下，選擇 AWS 服務。在選擇將使用此角色的服務下，選擇 Kinesis。在 Select your use case (選取您的使用案例) 下，選擇 Kinesis Analytics (Kinesis 分析)。

   選擇下一步：許可。

4. 在連接許可政策頁面，選擇下一步：檢閱。您會在建立角色後連接許可政策。

5. 在建立角色頁面，輸入 MF-stream-rw-role 作為角色名稱。選擇建立角色。

   現在您已建立名為 的新IAM角色MF-stream-rw-role。您接著會更新角色的信任和許可政策。

6. 將 許可政策連接到角色。

   注意

   在此練習中，Managed Service for Apache Flink 擔任從 Kinesis 資料串流 (來源) 讀取資料並將輸出寫入另一個 Kinesis 資料串流的角色。因此您會連接在上一個步驟中建立的政策，建立許可政策。

   1. 在摘要頁面，選擇許可標籤。

   2. 選擇連接政策。

   3. 在搜尋方塊中，輸入 AKReadSourceStreamWriteSinkStream (您在上一節中建立的政策)。

   4. 選擇AKReadSourceStreamWriteSinkStream政策，然後選擇附加政策 。

1. 將下列JSON程式碼儲存至名為 的檔案create_request.json。將範例角色取代ARN為您先前建立的角色ARN的 。將儲存貯體尾碼 ARN （username） 取代為您在上一節中選擇的尾碼。使用您的帳戶 ID 取代服務執行角色中的範例帳戶 ID (012345678901)。

   {
       "ApplicationName": "test",
       "ApplicationDescription": "my java test app",
       "RuntimeEnvironment": "FLINK-1_6",
       "ServiceExecutionRole": "arn:aws:iam::012345678901:role/MF-stream-rw-role",
       "ApplicationConfiguration": {
           "ApplicationCodeConfiguration": {
               "CodeContent": {
                   "S3ContentLocation": {
                       "BucketARN": "arn:aws:s3:::ka-app-code-username",
                       "FileKey": "java-getting-started-1.0.jar"
                   }
               },
               "CodeContentType": "ZIPFILE"
           },
           "EnvironmentProperties":  { 
            "PropertyGroups": [ 
               { 
                  "PropertyGroupId": "ProducerConfigProperties",
                  "PropertyMap" : {
                       "flink.stream.initpos" : "LATEST",
                       "aws.region" : "us-west-2",
                       "AggregationEnabled" : "false"
                  }
               },
               { 
                  "PropertyGroupId": "ConsumerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2"
                  }
               }
            ]
         }
       }
   }
   
   

2. 使用前述請求執行 CreateApplication 動作以建立應用程式：

   aws kinesisanalyticsv2 create-application --cli-input-json file://create_request.json

啟動應用程式

1. 將下列JSON程式碼儲存至名為 的檔案start_request.json。

   {
       "ApplicationName": "test",
       "RunConfiguration": {
           "ApplicationRestoreConfiguration": { 
            "ApplicationRestoreType": "RESTORE_FROM_LATEST_SNAPSHOT"
            }
       }
   }
   

2. 以啟動應用程式的上述請求，執行 StartApplication 動作：

   aws kinesisanalyticsv2 start-application --cli-input-json file://start_request.json

停止應用程式

1. 將下列JSON程式碼儲存至名為 的檔案stop_request.json。

   {
      "ApplicationName": "test"
   }
   

2. 以停止應用程式的上述請求，執行 StopApplication 動作：

   aws kinesisanalyticsv2 stop-application --cli-input-json file://stop_request.json

更新應用程式的環境屬性

1. 將下列JSON程式碼儲存至名為 的檔案update_properties_request.json。

   {"ApplicationName": "test",
      "CurrentApplicationVersionId": 1,
      "ApplicationConfigurationUpdate": { 
         "EnvironmentPropertyUpdates": { 
            "PropertyGroups": [ 
               { 
                  "PropertyGroupId": "ProducerConfigProperties",
                  "PropertyMap" : {
                       "flink.stream.initpos" : "LATEST",
                       "aws.region" : "us-west-2",
                       "AggregationEnabled" : "false"
                  }
               },
               { 
                  "PropertyGroupId": "ConsumerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2"
                  }
               }
            ]
         }
      }
   }
   

2. 使用前述請求執行 UpdateApplication 動作以更新環境屬性：

   aws kinesisanalyticsv2 update-application --cli-input-json file://update_properties_request.json

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

2. 在 Managed Service for Apache Flink 面板中，選擇 MyApplication。

3. 選擇設定。

4. 在快照區段中，選擇停用，然後選擇更新。

5. 在應用程式的頁面，選擇刪除，然後確認刪除。

1. 在 https://console.aws.amazon.com/flink 開啟 Managed Service for Apache Flink 主控台

2. 在 Kinesis Data Streams 面板中，選擇 ExampleInputStream。

3. 在 ExampleInputStream頁面中，選擇刪除 Kinesis Stream，然後確認刪除。

4. 在 Kinesis 串流頁面中，選擇 ExampleOutputStream，選擇動作 ，選擇刪除 ，然後確認刪除。

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

2. 選擇 ka-app-code-<username> 儲存貯體。

3. 選擇刪除，然後輸入儲存貯體名稱以確認刪除。

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

2. 在導覽列中，選擇政策。

3. 在篩選器控制項中，輸入 kinesis。

4. 選擇 kinesis-analytics-service-MyApplication-us-west-2 政策。

5. 選擇政策動作，然後選擇刪除。

6. 在導覽列中，選擇角色。

7. 選擇 kinesis-analytics-MyApplication-us-west-2 角色。

8. 選擇刪除角色，然後確認刪除。

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

2. 在導覽列中，選擇日誌。

3. 選擇 /aws/kinesis-analytics/MyApplication 日誌群組。

4. 選擇刪除日誌群組，然後確認刪除。

1. 使用下列內容建立名為 stock.py 的檔案：

   
       import datetime
       import json
       import random
       import boto3
   
       STREAM_NAME = "ExampleInputStream"
   
   
       def get_data():
           return {
               'event_time': datetime.datetime.now().isoformat(),
               'ticker': random.choice(['AAPL', 'AMZN', 'MSFT', 'INTC', 'TBV']),
               'price': round(random.random() * 100, 2)}
   
   
       def generate(stream_name, kinesis_client):
           while True:
               data = get_data()
               print(data)
               kinesis_client.put_record(
                   StreamName=stream_name,
                   Data=json.dumps(data),
                   PartitionKey="partitionkey")
   
   
       if __name__ == '__main__':
           generate(STREAM_NAME, boto3.client('kinesis', region_name='us-west-2'))

2. 執行 stock.py 指令碼：

   $ python stock.py

   在完成教學課程的其餘部分時，讓指令碼保持執行狀態。

1. 如果您尚未安裝 Git 用戶端，請先安裝。如需詳細資訊，請參閱安裝 Git。

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

   git clone https://github.com/aws-samples/amazon-kinesis-data-analytics-examples.git

3. 導覽至 amazon-kinesis-data-analytics-java-examples/TumblingWindow 目錄。

1. 如果尚未安裝 Java 和 Maven，請先安裝。如需詳細資訊，請參閱教學課程：開始使用 Managed Service for Apache Flink 中的 DataStream API 教學課程中的完成必要的先決條件。

2. 使用下列命令編譯應用程式：

   mvn package -Dflink.version=1.15.3

   注意

   提供的來源程式碼依賴於 Java 11 中的程式庫。

1. 在 Amazon S3 主控台中，選擇 ka-app-code-<username> 儲存貯體，然後選擇上傳 。

2. 在選取檔案步驟中，選擇 新增檔案。導覽至您在上一步驟中建立的 aws-kinesis-analytics-java-apps-1.0.jar 檔案。

3. 您不需要變更物件的任何設定，因此請選擇上傳。

1. 在 https://console.aws.amazon.com/flink 開啟 Managed Service for Apache Flink 主控台

2. 在 Managed Service for Apache Flink 儀表板上，選擇建立分析應用程式。

3. 在 Managed Service for Apache Flink - 建立應用程式頁面，提供應用程式詳細資訊，如下所示：

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

   • 對於​執行期，選擇 ​Apache Flink。

     注意

     Managed Service for Apache Flink 使用 Apache Flink 1.15.2 版。

   • 將版本下拉式清單保留為 Apache Flink 1.15.2 版 (建議版本)。

4. 針對存取許可 ，選擇建立/更新IAM角色 kinesis-analytics-MyApplication-us-west-2。

5. 選擇建立應用程式。

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

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

3. 在摘要頁面，選擇編輯政策。選擇 JSON 索引標籤。

4. 將下列政策範例的反白部分新增至政策。取代範例帳戶 IDs（012345678901） 搭配您的帳戶 ID。

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "logs:DescribeLogGroups",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*",
                   "arn:aws:s3:::ka-app-code-<username>/aws-kinesis-analytics-java-apps-1.0.jar"
               ]
           },
           {
               "Sid": "DescribeLogStreams",
               "Effect": "Allow",
               "Action": "logs:DescribeLogStreams",
               "Resource": "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:*"
           },
           {
               "Sid": "PutLogEvents",
               "Effect": "Allow",
               "Action": "logs:PutLogEvents",
               "Resource": "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:kinesis-analytics-log-stream"
           },
           {
               "Sid": "ListCloudwatchLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*"
               ]
           },
           {
               "Sid": "ReadInputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream"
           },
           {
               "Sid": "WriteOutputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleOutputStream"
           }
       ]
   }

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

2. 在設定應用程式頁面，提供程式碼位置：

   • 對於 Amazon S3 儲存貯體，請輸入 ka-app-code-<username>。

   • 對於 Amazon S3 物件的路徑，請輸入 aws-kinesis-analytics-java-apps-1.0.jar。

3. 在存取應用程式資源 下，針對存取許可 ，選擇建立/更新IAM角色 kinesis-analytics-MyApplication-us-west-2。

4. 在監控下，確保監控指標層級設為應用程式。

5. 對於CloudWatch 記錄 ，選取啟用核取方塊。

6. 選擇更新。

1. 在頁面上MyApplication，選擇執行 。保持選取不使用快照執行選項，然後確認動作。

2. 應用程式執行時，重新整理頁面。主控台會顯示 Application graph (應用程式圖形)。

1. 在 https://console.aws.amazon.com/flink 開啟 Managed Service for Apache Flink 主控台

2. 在 Managed Service for Apache Flink 面板中，選擇 MyApplication。

3. 在應用程式的頁面，選擇刪除，然後確認刪除。

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

2. 在 Kinesis Data Streams 面板中，選擇 ExampleInputStream。

3. 在ExampleInputStream頁面中，選擇刪除 Kinesis Stream，然後確認刪除。

4. 在 Kinesis 串流頁面中，選擇 ExampleOutputStream，選擇動作 ，選擇刪除 ，然後確認刪除。

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

2. 選擇 ka-app-code-<username> 儲存貯體。

3. 選擇刪除，然後輸入儲存貯體名稱以確認刪除。

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

2. 在導覽列中，選擇政策。

3. 在篩選器控制項中，輸入 kinesis。

4. 選擇 kinesis-analytics-service-MyApplication-us-west-2 政策。

5. 選擇政策動作，然後選擇刪除。

6. 在導覽列中，選擇角色。

7. 選擇 kinesis-analytics-MyApplication-us-west-2 角色。

8. 選擇刪除角色，然後確認刪除。

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

2. 在導覽列中，選擇日誌。

3. 選擇 /aws/kinesis-analytics/MyApplication 日誌群組。

4. 選擇刪除日誌群組，然後確認刪除。

1. 使用下列內容建立名為 stock.py 的檔案：

    
   import datetime
   import json
   import random
   import boto3
   
   STREAM_NAME = "ExampleInputStream"
   
   
   def get_data():
       return {
           "EVENT_TIME": datetime.datetime.now().isoformat(),
           "TICKER": random.choice(["AAPL", "AMZN", "MSFT", "INTC", "TBV"]),
           "PRICE": round(random.random() * 100, 2),
       }
   
   
   def generate(stream_name, kinesis_client):
       while True:
           data = get_data()
           print(data)
           kinesis_client.put_record(
               StreamName=stream_name, Data=json.dumps(data), PartitionKey="partitionkey"
           )
   
   
   if __name__ == "__main__":
       generate(STREAM_NAME, boto3.client("kinesis"))
   

2. 執行 stock.py 指令碼：

   $ python stock.py

   在完成教學課程的其餘部分時，讓指令碼保持執行狀態。

1. 如果您尚未安裝 Git 用戶端，請先安裝。如需詳細資訊，請參閱安裝 Git。

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

   git clone https://github.com/aws-samples/amazon-kinesis-data-analytics-examples.git

3. 導覽至 amazon-kinesis-data-analytics-java-examples/SlidingWindow 目錄。

1. 如果尚未安裝 Java 和 Maven，請先安裝。如需詳細資訊，請參閱教學課程：開始使用 Managed Service for Apache Flink 中的 DataStream API 教學課程中的完成必要的先決條件。

2. 使用下列命令編譯應用程式：

   mvn package -Dflink.version=1.15.3

   注意

   提供的來源程式碼依賴於 Java 11 中的程式庫。

1. 在 Amazon S3 主控台中，選擇 ka-app-code-<username> 儲存貯體，然後選擇上傳 。

2. 在選取檔案步驟中，選擇 新增檔案。導覽至您在上一步驟中建立的 aws-kinesis-analytics-java-apps-1.0.jar 檔案。

3. 您不需要變更物件的任何設定，因此請選擇上傳。

1. 在 https://console.aws.amazon.com/flink 開啟 Managed Service for Apache Flink 主控台

2. 在 Managed Service for Apache Flink 儀表板上，選擇建立分析應用程式。

3. 在 Managed Service for Apache Flink - 建立應用程式頁面，提供應用程式詳細資訊，如下所示：

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

   • 對於​執行期，選擇 ​Apache Flink。

   • 將版本下拉式清單保留為 Apache Flink 1.15.2 版 (建議版本)。

4. 針對存取許可 ，選擇建立/更新IAM角色 kinesis-analytics-MyApplication-us-west-2。

5. 選擇建立應用程式。

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

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

3. 在摘要頁面，選擇編輯政策。選擇 JSON 索引標籤。

4. 將下列政策範例的反白部分新增至政策。取代範例帳戶 IDs（012345678901） 搭配您的帳戶 ID。

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "logs:DescribeLogGroups",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*",
                   "arn:aws:s3:::ka-app-code-<username>/aws-kinesis-analytics-java-apps-1.0.jar"
               ]
           },
           {
               "Sid": "DescribeLogStreams",
               "Effect": "Allow",
               "Action": "logs:DescribeLogStreams",
               "Resource": "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:*"
           },
           {
               "Sid": "PutLogEvents",
               "Effect": "Allow",
               "Action": "logs:PutLogEvents",
               "Resource": "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:kinesis-analytics-log-stream"
           },
           {
               "Sid": "ListCloudwatchLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*"
               ]
           },
           {
               "Sid": "ReadInputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream"
           },
           {
               "Sid": "WriteOutputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleOutputStream"
           }
       ]
   }

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

2. 在設定應用程式頁面，提供程式碼位置：

   • 對於 Amazon S3 儲存貯體，請輸入 ka-app-code-<username>。

   • 對於 Amazon S3 物件的路徑，請輸入 aws-kinesis-analytics-java-apps-1.0.jar。

3. 在存取應用程式資源 下，針對存取許可 ，選擇建立/更新IAM角色 kinesis-analytics-MyApplication-us-west-2。

4. 在監控下，確保監控指標層級設為應用程式。

5. 對於CloudWatch 記錄 ，選取啟用核取方塊。

6. 選擇更新。

1. 在 https://console.aws.amazon.com/flink 開啟 Managed Service for Apache Flink 主控台

2. 在 Managed Service for Apache Flink 面板中，選擇 MyApplication。

3. 在應用程式的頁面，選擇刪除，然後確認刪除。

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

2. 在 Kinesis Data Streams 面板中，選擇 ExampleInputStream。

3. 在ExampleInputStream頁面中，選擇刪除 Kinesis Stream，然後確認刪除。

4. 在 Kinesis 串流頁面中，選擇 ExampleOutputStream，選擇動作 ，選擇刪除 ，然後確認刪除。

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

2. 選擇 ka-app-code-<username> 儲存貯體。

3. 選擇刪除，然後輸入儲存貯體名稱以確認刪除。

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

2. 在導覽列中，選擇政策。

3. 在篩選器控制項中，輸入 kinesis。

4. 選擇 kinesis-analytics-service-MyApplication-us-west-2 政策。

5. 選擇政策動作，然後選擇刪除。

6. 在導覽列中，選擇角色。

7. 選擇 kinesis-analytics-MyApplication-us-west-2 角色。

8. 選擇刪除角色，然後確認刪除。

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

2. 在導覽列中，選擇日誌。

3. 選擇 /aws/kinesis-analytics/MyApplication 日誌群組。

4. 選擇刪除日誌群組，然後確認刪除。

1. 使用下列內容建立名為 stock.py 的檔案：

   
       import datetime
       import json
       import random
       import boto3
   
       STREAM_NAME = "ExampleInputStream"
   
   
       def get_data():
           return {
               'event_time': datetime.datetime.now().isoformat(),
               'ticker': random.choice(['AAPL', 'AMZN', 'MSFT', 'INTC', 'TBV']),
               'price': round(random.random() * 100, 2)}
   
   
       def generate(stream_name, kinesis_client):
           while True:
               data = get_data()
               print(data)
               kinesis_client.put_record(
                   StreamName=stream_name,
                   Data=json.dumps(data),
                   PartitionKey="partitionkey")
   
   
       if __name__ == '__main__':
           generate(STREAM_NAME, boto3.client('kinesis', region_name='us-west-2'))

2. 執行 stock.py 指令碼：

   $ python stock.py

   在完成教學課程的其餘部分時，讓指令碼保持執行狀態。

1. 如果您尚未安裝 Git 用戶端，請先安裝。如需詳細資訊，請參閱安裝 Git。

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

   git clone https://github.com/aws-samples/amazon-kinesis-data-analytics-examples.git

3. 導覽至 amazon-kinesis-data-analytics-java-examples/S3Sink 目錄。

1. 如果尚未安裝 Java 和 Maven，請先安裝。如需詳細資訊，請參閱教學課程：開始使用 Managed Service for Apache Flink 中的 DataStream API 教學課程中的完成必要的先決條件。

2. 使用下列命令編譯應用程式：

   mvn package -Dflink.version=1.15.3

1. 在 Amazon S3 主控台中，選擇 ka-app-code-<username> 儲存貯體，導覽至程式碼資料夾，然後選擇上傳 。

2. 在選取檔案步驟中，選擇 新增檔案。導覽至您在上一步驟中建立的 aws-kinesis-analytics-java-apps-1.0.jar 檔案。

3. 您不需要變更物件的任何設定，因此請選擇上傳。

1. 在 https://console.aws.amazon.com/flink 開啟 Managed Service for Apache Flink 主控台

2. 在 Managed Service for Apache Flink 儀表板上，選擇建立分析應用程式。

3. 在 Managed Service for Apache Flink - 建立應用程式頁面，提供應用程式詳細資訊，如下所示：

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

   • 對於​執行期，選擇 ​Apache Flink。

   • 將版本下拉式清單保留為 Apache Flink 1.15.2 版 (建議版本)。

4. 針對存取許可 ，選擇建立/更新IAM角色 kinesis-analytics-MyApplication-us-west-2。

5. 選擇建立應用程式。

   注意

   當您使用主控台建立 Managed Service for Apache Flink 應用程式時，您可以選擇為應用程式建立IAM角色和政策。應用程式使用此角色和政策來存取其相依資源。這些IAM資源會使用您的應用程式名稱和區域命名，如下所示：

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

   • 對於​執行期，選擇 ​Apache Flink。

   • 將版本保留為 Apache Flink 1.15.2 版 (建議版本)。

6. 針對存取許可 ，選擇建立/更新IAM角色 kinesis-analytics-MyApplication-us-west-2。

7. 選擇建立應用程式。

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

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

3. 在摘要頁面，選擇編輯政策。選擇 JSON 索引標籤。

4. 將下列政策範例的反白部分新增至政策。取代範例帳戶 IDs（012345678901） 搭配您的帳戶 ID。使用您的使用者名稱取代 <username>。

   {
               "Sid": "S3",
               "Effect": "Allow",
               "Action": [
                   "s3:Abort*",
                   "s3:DeleteObject*",
                   "s3:GetObject*",
                   "s3:GetBucket*",
                   "s3:List*",
                   "s3:ListBucket",
                   "s3:PutObject"
               ],
               "Resource": [
                   "arn:aws:s3:::ka-app-code-<username>",
                   "arn:aws:s3:::ka-app-code-<username>/*"
               ]
             }, 
           {
               "Sid": "ListCloudwatchLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:region:account-id:log-group:*"
               ]
           },
           {
               "Sid": "ListCloudwatchLogStreams",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogStreams"
               ],
               "Resource": [
                   "arn:aws:logs:region:account-id:log-group:%LOG_GROUP_PLACEHOLDER%:log-stream:*"
               ]
           },
           {
               "Sid": "PutCloudwatchLogs",
               "Effect": "Allow",
               "Action": [
                   "logs:PutLogEvents"
               ],
               "Resource": [
                   "arn:aws:logs:region:account-id:log-group:%LOG_GROUP_PLACEHOLDER%:log-stream:%LOG_STREAM_PLACEHOLDER%"
               ]
           }
           ,
           {
               "Sid": "ReadInputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream"
           },
           
       ]
   }

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

2. 在設定應用程式頁面，提供程式碼位置：

   • 對於 Amazon S3 儲存貯體，請輸入 ka-app-code-<username>。

   • 對於 Amazon S3 物件的路徑，請輸入 code/aws-kinesis-analytics-java-apps-1.0.jar。

3. 在存取應用程式資源 下，針對存取許可 ，選擇建立/更新IAM角色 kinesis-analytics-MyApplication-us-west-2。

4. 在監控下，確保監控指標層級設為應用程式。

5. 對於CloudWatch 記錄 ，選取啟用核取方塊。

6. 選擇更新。

1. 在頁面上MyApplication，選擇執行 。保持選取不使用快照執行選項，然後確認動作。

2. 應用程式執行時，重新整理頁面。主控台會顯示 Application graph (應用程式圖形)。

1. 將以下 import 陳述式新增到 S3StreamingSinkJob.java 檔案的開頭：

   import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
   import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.DateTimeBucketAssigner;

2. 將程式碼中的 createS3SinkFromStaticConfig() 方法更新為如下所示：

   private static StreamingFileSink<String> createS3SinkFromStaticConfig() {
   
       final StreamingFileSink<String> sink = StreamingFileSink
           .forRowFormat(new Path(s3SinkPath), new SimpleStringEncoder<String>("UTF-8"))
           .withBucketAssigner(new DateTimeBucketAssigner("yyyy-MM-dd--HH"))
           .withRollingPolicy(DefaultRollingPolicy.create().build())
           .build();
       return sink;
   }
               

1. 增加應用程式的 CheckpointInterval 設定。 UpdateApplication 動作的下列輸入會將檢查點間隔設定為 10 分鐘：

   {
      "ApplicationConfigurationUpdate": {  
         "FlinkApplicationConfigurationUpdate": { 
            "CheckpointConfigurationUpdate": {
               "ConfigurationTypeUpdate" : "CUSTOM", 
               "CheckpointIntervalUpdate": 600000
            }
         }
      },
      "ApplicationName": "MyApplication",
      "CurrentApplicationVersionId": 5
   }

   若要使用上述程式碼，請指定目前的應用程式版本。您可以使用 ListApplications動作來擷取應用程式版本。

2. 將以下 import 陳述式新增到 S3StreamingSinkJob.java 檔案的開頭：

   import java.util.concurrent.TimeUnit; 

3. 將 S3StreamingSinkJob.java 檔案中的 createS3SinkFromStaticConfig 方法更新為如下所示：

   private static StreamingFileSink<String> createS3SinkFromStaticConfig() {
   
           final StreamingFileSink<String> sink = StreamingFileSink
                   .forRowFormat(new Path(s3SinkPath), new SimpleStringEncoder<String>("UTF-8"))
                   .withBucketAssigner(new DateTimeBucketAssigner("yyyy-MM-dd--HH"))
                   .withRollingPolicy(
                           DefaultRollingPolicy.create()
                               .withRolloverInterval(TimeUnit.MINUTES.toMillis(8))
                               .withInactivityInterval(TimeUnit.MINUTES.toMillis(5))
                               .withMaxPartSize(1024 * 1024 * 1024)
                               .build())
                   .build();
           return sink;
       }

   上述程式碼範例會將寫入 Amazon S3 儲存貯體的頻率設定為 8 分鐘。

1. 在 https://console.aws.amazon.com/flink 開啟 Managed Service for Apache Flink 主控台

2. 在 Managed Service for Apache Flink 面板中，選擇 MyApplication。

3. 在應用程式頁面，選擇刪除，然後確認刪除。

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

2. 在 Kinesis Data Streams 面板中，選擇 ExampleInputStream。

3. 在ExampleInputStream頁面上，選擇刪除 Kinesis Stream，然後確認刪除。

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

2. 選擇 ka-app-code-<username> 儲存貯體。

3. 選擇刪除，然後輸入儲存貯體名稱以確認刪除。

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

2. 在導覽列中，選擇政策。

3. 在篩選器控制項中，輸入 kinesis。

4. 選擇 kinesis-analytics-service-MyApplication-us-west-2 政策。

5. 選擇政策動作，然後選擇刪除。

6. 在導覽列中，選擇角色。

7. 選擇 kinesis-analytics-MyApplication-us-west-2 角色。

8. 選擇刪除角色，然後確認刪除。

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

2. 在導覽列中，選擇日誌。

3. 選擇 /aws/kinesis-analytics/MyApplication 日誌群組。

4. 選擇刪除日誌群組，然後確認刪除。

1. 如果您尚未安裝 Git 用戶端，請先安裝。如需詳細資訊，請參閱安裝 Git。

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

   git clone https://github.com/aws-samples/amazon-kinesis-data-analytics-examples.git

3. 應用程式的程式碼位於 amazon-kinesis-data-analytics-java-examples/KafkaConnectors/KafkaGettingStartedJob.java 檔案中。您可以檢查程式碼，以熟悉 Managed Service for Apache Flink 應用程式程式碼的結構。

4. 使用命令列 Maven 工具或偏好的開發環境來建立JAR檔案。若要使用命令列 Maven 工具編譯JAR檔案，請輸入下列內容：

   mvn package -Dflink.version=1.15.3

   如果建置成功，會建立下列檔案：

   target/KafkaGettingStartedJob-1.0.jar

   注意

   提供的來源程式碼依賴於 Java 11 中的程式庫。如果您使用的是開發環境，

1. 在 Amazon S3 主控台中，選擇 ka-app-code-<username> 儲存貯體，然後選擇上傳 。

2. 在選取檔案步驟中，選擇 新增檔案。導覽至您在上一步驟中建立的 KafkaGettingStartedJob-1.0.jar 檔案。

3. 您不需要變更物件的任何設定，因此請選擇上傳。

1. 在 https://console.aws.amazon.com/flink 開啟 Managed Service for Apache Flink 主控台。

2. 在 Managed Service for Apache Flink 儀表板上，選擇建立分析應用程式。

3. 在 Managed Service for Apache Flink - 建立應用程式頁面，提供應用程式詳細資訊，如下所示：

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

   • 針對​執行期，選擇 ​Apache Flink 1.15.2 版。

4. 針對存取許可 ，選擇建立/更新IAM角色 kinesis-analytics-MyApplication-us-west-2。

5. 選擇建立應用程式。

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

2. 在設定應用程式頁面，提供程式碼位置：

   • 對於 Amazon S3 儲存貯體，請輸入 ka-app-code-<username>。

   • 對於 Amazon S3 物件的路徑，請輸入 KafkaGettingStartedJob-1.0.jar。

3. 在存取應用程式資源 下，針對存取許可 ，選擇建立/更新IAM角色 kinesis-analytics-MyApplication-us-west-2。

   注意

   當您使用主控台 （例如 CloudWatch Logs 或 Amazon VPC） 指定應用程式資源時，主控台會修改您的應用程式執行角色，以授予存取這些資源的許可。

4. 在屬性下，選擇新增群組。輸入下列屬性：

   群組 ID	金鑰	值
   KafkaSource	主題	AWS KafkaTutorialTopic
   KafkaSource	bootstrap.servers	The bootstrap server list you saved previously
   KafkaSource	security.protocol	SSL
   KafkaSource	ssl.truststore.location	/usr/lib/jvm/java-11-amazon-corretto/lib/security/cacerts
   KafkaSource	ssl.truststore.password	changeit

   注意

   預設憑證的 ssl.truststore.password 是「changeit」；如果您使用預設憑證，不需要變更此值。

   再次選擇新增群組。輸入下列屬性：

   群組 ID	金鑰	值
   KafkaSink	主題	AWS KafkaTutorialTopicDestination
   KafkaSink	bootstrap.servers	The bootstrap server list you saved previously
   KafkaSink	security.protocol	SSL
   KafkaSink	ssl.truststore.location	/usr/lib/jvm/java-11-amazon-corretto/lib/security/cacerts
   KafkaSink	ssl.truststore.password	changeit
   KafkaSink	transaction.timeout.ms	1000

   應用程式程式碼會讀取上述應用程式屬性，以設定用於與您的 和 Amazon MSK叢集互動的來源VPC和接收。如需如何使用屬性的詳細資訊，請參閱在 Apache Flink 的受管理服務中使用執行階段屬性。

5. 在快照下選擇停用。這可以讓您更輕鬆地更新應用程式，而無需加載無效的應用程式狀態資料。

6. 在監控下，確保監控指標層級設為應用程式。

7. 對於CloudWatch 記錄 ，請選擇啟用核取方塊。

8. 在虛擬私有雲端 （VPC） 區段中，選擇要與您的應用程式VPC建立關聯的 。選擇與 相關聯的子網路和安全群組VPC，讓應用程式用來存取VPC資源。

9. 選擇更新。

1. 使用下列內容建立名為 stock.py 的檔案：

    
       import datetime
       import json
       import random
       import boto3
   
       STREAM_NAME = "ExampleInputStream"
   
   
       def get_data():
           return {
               'event_time': datetime.datetime.now().isoformat(),
               'ticker': random.choice(['AAPL', 'AMZN', 'MSFT', 'INTC', 'TBV']),
               'price': round(random.random() * 100, 2)}
   
   
       def generate(stream_name, kinesis_client):
           while True:
               data = get_data()
               print(data)
               kinesis_client.put_record(
                   StreamName=stream_name,
                   Data=json.dumps(data),
                   PartitionKey="partitionkey")
   
   
       if __name__ == '__main__':
           generate(STREAM_NAME, boto3.client('kinesis', region_name='us-west-2'))

2. 執行 stock.py 指令碼：

   $ python stock.py

   在完成教學課程的其餘部分時，讓指令碼保持執行狀態。

1. 如果您尚未安裝 Git 用戶端，請先安裝。如需詳細資訊，請參閱安裝 Git。

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

   git clone https://github.com/aws-samples/amazon-kinesis-data-analytics-examples.git

3. 導覽至 amazon-kinesis-data-analytics-java-examples/EfoConsumer 目錄。

1. 如果尚未安裝 Java 和 Maven，請先安裝。如需詳細資訊，請參閱教學課程：開始使用 Managed Service for Apache Flink 中的 DataStream API 教學課程中的完成必要的先決條件。

2. 使用下列命令編譯應用程式：

   mvn package -Dflink.version=1.15.3

   注意

   提供的來源程式碼依賴於 Java 11 中的程式庫。

1. 在 Amazon S3 主控台中，選擇 ka-app-code-<username> 儲存貯體，然後選擇上傳 。

2. 在選取檔案步驟中，選擇 新增檔案。導覽至您在上一步驟中建立的 aws-kinesis-analytics-java-apps-1.0.jar 檔案。

3. 您不需要變更物件的任何設定，因此請選擇上傳。

1. 在 https://console.aws.amazon.com/flink 開啟 Managed Service for Apache Flink 主控台

2. 在 Managed Service for Apache Flink 儀表板上，選擇建立分析應用程式。

3. 在 Managed Service for Apache Flink - 建立應用程式頁面，提供應用程式詳細資訊，如下所示：

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

   • 對於​執行期，選擇 ​Apache Flink。

     注意

     Managed Service for Apache Flink 使用 Apache Flink 1.15.2 版。

   • 將版本下拉式清單保留為 Apache Flink 1.15.2 版 (建議版本)。

4. 針對存取許可 ，選擇建立/更新IAM角色 kinesis-analytics-MyApplication-us-west-2。

5. 選擇建立應用程式。

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

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

3. 在摘要頁面，選擇編輯政策。選擇 JSON 索引標籤。

4. 將下列政策範例的反白部分新增至政策。取代範例帳戶 IDs（012345678901） 搭配您的帳戶 ID。

   注意

   這些許可授予應用程式存取EFO取用者的能力。

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "logs:DescribeLogGroups",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*",
                   "arn:aws:s3:::ka-app-code-<username>/aws-kinesis-analytics-java-apps-1.0.jar"
               ]
           },
           {
               "Sid": "DescribeLogStreams",
               "Effect": "Allow",
               "Action": "logs:DescribeLogStreams",
               "Resource": "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:*"
           },
           {
               "Sid": "PutLogEvents",
               "Effect": "Allow",
               "Action": "logs:PutLogEvents",
               "Resource": "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:kinesis-analytics-log-stream"
           },
           {
               "Sid": "ListCloudwatchLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*"
               ]
           },
           {
               "Sid": "AllStreams",
               "Effect": "Allow",
               "Action": [
                   "kinesis:ListShards",
                   "kinesis:ListStreamConsumers",
                   "kinesis:DescribeStreamSummary"
               ],
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/*"
           },
           {
               "Sid": "Stream",
               "Effect": "Allow",
               "Action": [
                   "kinesis:DescribeStream",
                   "kinesis:RegisterStreamConsumer",
                   "kinesis:DeregisterStreamConsumer"
               ],
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream"
           },
           {
               "Sid": "WriteOutputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleOutputStream"
           },
           {
               "Sid": "Consumer",
               "Effect": "Allow",
               "Action": [
                   "kinesis:DescribeStreamConsumer",
                   "kinesis:SubscribeToShard"
               ],
               "Resource": [
                   "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream/consumer/my-efo-flink-app",
                   "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream/consumer/my-efo-flink-app:*"
               ]
           }
       ]
   }

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

2. 在設定應用程式頁面，提供程式碼位置：

   • 對於 Amazon S3 儲存貯體，請輸入 ka-app-code-<username>。

   • 對於 Amazon S3 物件的路徑，請輸入 aws-kinesis-analytics-java-apps-1.0.jar。

3. 在存取應用程式資源 下，針對存取許可 ，選擇建立/更新IAM角色 kinesis-analytics-MyApplication-us-west-2。

4. 在屬性下，選擇建立群組。

5. 輸入以下應用程式屬性和數值：

   群組 ID	金鑰	值
   ConsumerConfigProperties	flink.stream.recordpublisher	EFO
   ConsumerConfigProperties	flink.stream.efo.consumername	basic-efo-flink-app
   ConsumerConfigProperties	INPUT_STREAM	ExampleInputStream
   ConsumerConfigProperties	flink.inputstream.initpos	LATEST
   ConsumerConfigProperties	AWS_REGION	us-west-2

6. 在屬性下，選擇建立群組。

7. 輸入以下應用程式屬性和數值：

   群組 ID	金鑰	值
   ProducerConfigProperties	OUTPUT_STREAM	ExampleOutputStream
   ProducerConfigProperties	AWS_REGION	us-west-2
   ProducerConfigProperties	AggregationEnabled	false

8. 在監控下，確保監控指標層級設為應用程式。

9. 對於CloudWatch 記錄 ，選取啟用核取方塊。

10. 選擇更新。

1. 在 https://console.aws.amazon.com/flink 開啟 Managed Service for Apache Flink 主控台

2. 在 Managed Service for Apache Flink 面板中，選擇 MyApplication。

3. 在應用程式的頁面，選擇刪除，然後確認刪除。

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

2. 在 Kinesis Data Streams 面板中，選擇 ExampleInputStream。

3. 在ExampleInputStream頁面中，選擇刪除 Kinesis Stream，然後確認刪除。

4. 在 Kinesis 串流頁面中，選擇 ExampleOutputStream，選擇動作 ，選擇刪除 ，然後確認刪除。

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

2. 選擇 ka-app-code-<username> 儲存貯體。

3. 選擇刪除，然後輸入儲存貯體名稱以確認刪除。

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

2. 在導覽列中，選擇政策。

3. 在篩選器控制項中，輸入 kinesis。

4. 選擇 kinesis-analytics-service-MyApplication-us-west-2 政策。

5. 選擇政策動作，然後選擇刪除。

6. 在導覽列中，選擇角色。

7. 選擇 kinesis-analytics-MyApplication-us-west-2 角色。

8. 選擇刪除角色，然後確認刪除。

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

2. 在導覽列中，選擇日誌。

3. 選擇 /aws/kinesis-analytics/MyApplication 日誌群組。

4. 選擇刪除日誌群組，然後確認刪除。

1. 使用下列內容建立名為 stock.py 的檔案：

    import datetime
       import json
       import random
       import boto3
   
       STREAM_NAME = "ExampleInputStream"
   
   
       def get_data():
           return {
               'event_time': datetime.datetime.now().isoformat(),
               'ticker': random.choice(['AAPL', 'AMZN', 'MSFT', 'INTC', 'TBV']),
               'price': round(random.random() * 100, 2)}
   
   
       def generate(stream_name, kinesis_client):
           while True:
               data = get_data()
               print(data)
               kinesis_client.put_record(
                   StreamName=stream_name,
                   Data=json.dumps(data),
                   PartitionKey="partitionkey")
   
   
       if __name__ == '__main__':
           generate(STREAM_NAME, boto3.client('kinesis', region_name='us-west-2'))

2. 執行 stock.py 指令碼：

   $ python stock.py

   在完成教學課程的其餘部分時，讓指令碼保持執行狀態。

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

   git clone https://github.com/aws-samples/amazon-kinesis-data-analytics-examples.git

2. 導覽至 amazon-kinesis-data-analytics-java-examples/FirehoseSink 目錄。

1. 如果尚未安裝 Java 和 Maven，請先安裝。如需詳細資訊，請參閱教學課程：開始使用 Managed Service for Apache Flink 中的 DataStream API 教學課程中的完成必要的先決條件。

2. 若要將 Kinesis 連接器用於以下應用程式，您需要下載、建置並安裝 Apache Maven。如需詳細資訊，請參閱 將 Apache Flink Kinesis Streams 連接器與先前的 Apache Flink 版本搭配使用。

3. 使用下列命令編譯應用程式：

   mvn package -Dflink.version=1.15.3

   注意

   提供的來源程式碼依賴於 Java 11 中的程式庫。

上傳應用程式的程式碼

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

2. 在主控台中，選擇 ka-app-code-<username> 儲存貯體，然後選擇上傳 。

3. 在選取檔案步驟中，選擇 新增檔案。導覽至您在上一步驟中建立的 java-getting-started-1.0.jar 檔案。

4. 您不需要變更物件的任何設定，因此請選擇上傳。

1. 在 https://console.aws.amazon.com/flink 開啟 Managed Service for Apache Flink 主控台

2. 在 Managed Service for Apache Flink 儀表板上，選擇建立分析應用程式。

3. 在 Managed Service for Apache Flink - 建立應用程式頁面，提供應用程式詳細資訊，如下所示：

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

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

   • 對於​執行期，選擇 ​Apache Flink。

     注意

     Managed Service for Apache Flink 使用 Apache Flink 1.15.2 版。

   • 將版本下拉式清單保留為 Apache Flink 1.15.2 版 (建議版本)。

4. 針對存取許可 ，選擇建立/更新IAM角色 kinesis-analytics-MyApplication-us-west-2。

5. 選擇建立應用程式。

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

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

3. 在摘要頁面，選擇編輯政策。選擇 JSON 索引標籤。

4. 將下列政策範例的反白部分新增至政策。取代範例帳戶的所有執行個體 IDs（012345678901） 搭配您的帳戶 ID。

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:s3:::ka-app-code-username/java-getting-started-1.0.jar"
               ]
           },
           {
               "Sid": "DescribeLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*"
               ]
           },
           {
               "Sid": "DescribeLogStreams",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogStreams"
              ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:*"
               ]
           },
           {
               "Sid": "PutLogEvents",
               "Effect": "Allow",
               "Action": [
                   "logs:PutLogEvents"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:kinesis-analytics-log-stream"
               ]
           },
           {
               "Sid": "ReadInputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream"
           },
           {
               "Sid": "WriteDeliveryStream",
               "Effect": "Allow",
               "Action": "firehose:*",
               "Resource": "arn:aws:firehose:us-west-2:012345678901:deliverystream/ExampleDeliveryStream"
           }
       ]
   }

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

2. 在設定應用程式頁面，提供程式碼位置：

   • 對於 Amazon S3 儲存貯體，請輸入 ka-app-code-<username>。

   • 對於 Amazon S3 物件的路徑，請輸入 java-getting-started-1.0.jar。

3. 在存取應用程式資源 下，針對存取許可 ，選擇建立/更新IAM角色 kinesis-analytics-MyApplication-us-west-2。

4. 在監控下，確保監控指標層級設為應用程式。

5. 對於CloudWatch 記錄 ，選取啟用核取方塊。

6. 選擇更新。

建立 IAM 角色

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

2. 在導覽窗格中，選擇角色 、建立角色。

3. 在選取可信身分類型下，選擇 AWS 服務。在選擇將使用此角色的服務下，選擇 Kinesis。在 Select your use case (選取您的使用案例) 下，選擇 Kinesis Analytics (Kinesis 分析)。

   選擇下一步：許可。

4. 在連接許可政策頁面，選擇下一步：檢閱。您會在建立角色後連接許可政策。

5. 在建立角色頁面，輸入 MF-stream-rw-role 作為角色名稱。選擇建立角色。

   現在您已建立名為 的新IAM角色MF-stream-rw-role。您接著會更新角色的信任和許可政策。

6. 將 許可政策連接到角色。

   注意

   在此練習中，Managed Service for Apache Flink 擔任從 Kinesis 資料串流 (來源) 讀取資料並將輸出寫入另一個 Kinesis 資料串流的角色。因此您會連接在上一個步驟中建立的政策，建立許可政策。

   1. 在摘要頁面，選擇許可標籤。

   2. 選擇連接政策。

   3. 在搜尋方塊中，輸入 AKReadSourceStreamWriteSinkStream (您在上一節中建立的政策)。

   4. 選擇AKReadSourceStreamWriteSinkStream政策，然後選擇附加政策 。

1. 將下列JSON程式碼儲存至名為 的檔案create_request.json。將範例角色取代ARN為您先前建立的角色ARN的 。將儲存貯體ARN尾碼取代為您在 建立相依資源章節 （.） 中選擇的尾碼ka-app-code-<username>。取代範例帳戶 ID （012345678901） 位於服務執行角色中，並提供您的帳戶 ID。

   {
       "ApplicationName": "test",
       "ApplicationDescription": "my java test app",
       "RuntimeEnvironment": "FLINK-1_15",
       "ServiceExecutionRole": "arn:aws:iam::012345678901:role/MF-stream-rw-role",
       "ApplicationConfiguration": {
           "ApplicationCodeConfiguration": {
               "CodeContent": {
                   "S3ContentLocation": {
                       "BucketARN": "arn:aws:s3:::ka-app-code-username",
                       "FileKey": "java-getting-started-1.0.jar"
                   }
               },
               "CodeContentType": "ZIPFILE"
           }
         }
       }
   }
   
   

2. 使用前述請求執行 CreateApplication 動作以建立應用程式：

   aws kinesisanalyticsv2 create-application --cli-input-json file://create_request.json

啟動應用程式

1. 將下列JSON程式碼儲存至名為 的檔案start_request.json。

   {
       "ApplicationName": "test",
       "RunConfiguration": {
           "ApplicationRestoreConfiguration": { 
            "ApplicationRestoreType": "RESTORE_FROM_LATEST_SNAPSHOT"
            }
       }
   }
   

2. 以啟動應用程式的上述請求，執行 StartApplication 動作：

   aws kinesisanalyticsv2 start-application --cli-input-json file://start_request.json

停止應用程式

1. 將下列JSON程式碼儲存至名為 的檔案stop_request.json。

   
   {
       "ApplicationName": "test"
   }
   

2. 以停止應用程式的上述請求，執行 StopApplication 動作：

   aws kinesisanalyticsv2 stop-application --cli-input-json file://stop_request.json

1. 在 https://console.aws.amazon.com/flink 開啟 Managed Service for Apache Flink 主控台

2. 在 Managed Service for Apache Flink 面板中，選擇 MyApplication。

3. 選擇設定。

4. 在快照區段中，選擇停用，然後選擇更新。

5. 在應用程式的頁面，選擇刪除，然後確認刪除。

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

2. 在 Kinesis Data Streams 面板中，選擇 ExampleInputStream。

3. 在 ExampleInputStream頁面中，選擇刪除 Kinesis Stream，然後確認刪除。

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

2. 在 Firehose 面板中，選擇 ExampleDeliveryStream。

3. 在ExampleDeliveryStream頁面中，選擇刪除 Firehose 串流，然後確認刪除。

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

2. 選擇 ka-app-code-<username> 儲存貯體。

3. 選擇刪除，然後輸入儲存貯體名稱以確認刪除。

4. 如果您為 Firehose 串流的目的地建立 Amazon S3 儲存貯體，請同時刪除該儲存貯體。

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

2. 在導覽列中，選擇政策。

3. 在篩選器控制項中，輸入 kinesis。

4. 選擇 kinesis-analytics-service-MyApplication-us-west-2 政策。

5. 選擇政策動作，然後選擇刪除。

6. 如果您為 Firehose 串流建立新的政策，請同時刪除該政策。

7. 在導覽列中，選擇角色。

8. 選擇 kinesis-analytics-MyApplication-us-west-2 角色。

9. 選擇刪除角色，然後確認刪除。

10. 如果您為 Firehose 串流建立新的角色，請同時刪除該角色。

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

2. 在導覽列中，選擇日誌。

3. 選擇 /aws/kinesis-analytics/MyApplication 日誌群組。

4. 選擇刪除日誌群組，然後確認刪除。