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 Windows で Flink をコンパイルする場合は、 -Drat.skip=true パラメータを追加する必要があります。

にサインアップするには AWS アカウント

1. https://portal.aws.amazon.com/billing/サインアップ を開きます。

2. オンラインの手順に従います。

   サインアップ手順の一環として、通話呼び出しを受け取り、電話キーパッドで検証コードを入力するように求められます。

   にサインアップすると AWS アカウント、 AWS アカウントのルートユーザー が作成されます。ルートユーザーには、アカウントのすべての AWS のサービス とリソースへのアクセス権があります。セキュリティのベストプラクティスとして、ユーザーに管理アクセスを割り当て、ルートユーザーのみを使用してルートユーザーアクセスが必要なタスクを実行してください。

のセキュリティ保護 AWS アカウントのルートユーザー

1. ルートユーザーを選択し、 AWS アカウント E メールアドレスを入力して、アカウント所有者AWS Management Consoleとして にサインインします。次のページでパスワードを入力します。

   ルートユーザーを使用してサインインする方法については、AWS サインイン ユーザーガイドのルートユーザーとしてサインインするを参照してください。

2. ルートユーザーの多要素認証 (MFA) を有効にします。

   手順については、「 ユーザーガイド」の AWS アカウント 「ルートユーザー (コンソール) の仮想MFAデバイスの有効化」を参照してください。 IAM

管理アクセスを持つユーザーを作成する

1. IAM Identity Center を有効にします。

   手順については、「AWS IAM Identity Center ユーザーガイド」の「AWS IAM Identity Centerの有効化」を参照してください。

2. IAM Identity Center で、ユーザーに管理アクセスを許可します。

   を ID ソース IAM アイデンティティセンターディレクトリ として使用する方法のチュートリアルについては、 AWS IAM Identity Center ユーザーガイドの「デフォルトを使用してユーザーアクセスを設定する IAM アイデンティティセンターディレクトリ」を参照してください。

管理アクセス権を持つユーザーとしてサインインする

• IAM Identity Center ユーザーでサインインするには、IAMIdentity Center ユーザーの作成時に E メールアドレスに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. 管理者ユーザーの名前付きプロファイルを config ファイルに追加します AWS CLI 。 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. コマンドプロンプトで以下のヘルプコマンドを入力して、セットアップを確認します。

   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. アプリケーションが出力の書き込みに使用する 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ファイルにパッケージ化します。コードのコンパイルとパッケージ化には次の 2 通りの方法があります。

   • コマンドライン 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. [Bucket name (バケット名)] フィールドにka-app-code-<username>と入力します。バケット名にユーザー名などのサフィックスを追加して、グローバルに一意にします。[Next (次へ)] を選択します。

4. 設定オプションのステップでは、設定をそのままにし、[次へ] を選択します。

5. アクセス許可の設定のステップでは、設定をそのままにし、[次へ] を選択します。

6. [バケットを作成] を選択します。

7. Amazon S3 コンソールで、 ka-app-code- を選択します。<username> バケットを選択し、アップロード を選択します。

8. ファイルの選択のステップで、[ファイルを追加] を選択します。前のステップで作成した aws-kinesis-analytics-java-apps-1.0.jar ファイルに移動します。[Next (次へ)] を選択します。

9. オブジェクトの設定を変更する必要はないので、[アップロード] を選択してください。

1. Managed Service for Apache Flink コンソールを https://console.aws.amazon.com/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. [Create application] を選択します。

1. でIAMコンソールを開きますhttps://console.aws.amazon.com/iam/。

2. [Policies] (ポリシー) を選択します。前のセクションでコンソールによって作成された 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. [Configure application] ページで、[Code location] を次のように指定します。

   • [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. [Monitoring] の [Monitoring metrics level] が [Application] に設定されていることを確認します。

6. CloudWatch をログに記録するには、有効化チェックボックスをオンにします。

7. [Update] (更新) を選択します。

IAM ロールを作成するには

1. でIAMコンソールを開きますhttps://console.aws.amazon.com/iam/。

2. ナビゲーションペインで [Roles (ロール)]、[Create Role (ロールの作成)] の順に選択します。

3. [信頼されるエンティティの種類を選択] で、[AWS のサービス] を選択します。[このロールを使用するサービスを選択] で、[Kinesis Analytics] を選択します。[ユースケースの選択] で、[Kinesis Analytics ] を選択します。

   [Next: Permissions] (次へ: アクセス許可) を選択します。

4. [アクセス権限ポリシーをアタッチする] ページで、[Next: Review] (次: 確認) を選択します。ロールを作成した後に、アクセス許可ポリシーをアタッチします。

5. [Create role (ロールの作成)] ページで、ロールの名前に MF-stream-rw-role を入力します。[ロールの作成] を選択します。

   これで、 という新しいIAMロールが作成されましたMF-stream-rw-role。次に、ロールの信頼ポリシーとアクセス許可ポリシーを更新します。

6. アクセス許可ポリシーをロールにアタッチします。

   注記

   この演習では、Managed Service for Apache Flink が、Kinesis データストリーム (ソース) からのデータの読み取りと、別の Kinesis データストリームへの出力の書き込みの両方を実行するためにこのロールを引き受けます。このため、前のステップで作成したポリシー、許可ポリシーを作成する をアタッチします。

   1. [概要] ページで、[アクセス許可] タブを選択します。

   2. [Attach Policies (ポリシーのアタッチ)] を選択します。

   3. 検索ボックスにAKReadSourceStreamWriteSinkStream(前のセクションで作成したポリシー) と入力します。

   4. AKReadSourceStreamWriteSinkStream ポリシーを選択し、ポリシーのアタッチ を選択します。

1. 次のJSONコードを という名前のファイルに保存しますcreate_request.json。サンプルロールを、以前に作成したロールARNの ARNに置き換えます。バケットARNサフィックス (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": "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. Managed Service for Apache Flink コンソールを https://console.aws.amazon.com/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 アカウント E メールアドレスを入力して、アカウント所有者AWS Management Consoleとして にサインインします。次のページでパスワードを入力します。

   ルートユーザーを使用してサインインする方法については、AWS サインイン ユーザーガイドのルートユーザーとしてサインインするを参照してください。

2. ルートユーザーの多要素認証 (MFA) を有効にします。

   手順については、「 ユーザーガイド」の AWS アカウント 「ルートユーザー (コンソール) の仮想MFAデバイスの有効化」を参照してください。 IAM

管理アクセスを持つユーザーを作成する

1. IAM Identity Center を有効にします。

   手順については、「AWS IAM Identity Center ユーザーガイド」の「AWS IAM Identity Centerの有効化」を参照してください。

2. IAM Identity Center で、ユーザーに管理アクセスを許可します。

   ID ソース IAM アイデンティティセンターディレクトリ として を使用するためのチュートリアルについては、 AWS IAM Identity Center ユーザーガイドの「デフォルトを使用してユーザーアクセスを設定する IAM アイデンティティセンターディレクトリ」を参照してください。

管理アクセス権を持つユーザーとしてサインインする

• IAM Identity Center ユーザーでサインインするには、IAMIdentity Center ユーザーの作成時に E メールアドレスに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. 管理者ユーザーの名前付きプロファイルを config ファイルに追加します AWS CLI 。 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. コマンドプロンプトで以下のヘルプコマンドを入力して、セットアップを確認します。

   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. アプリケーションが出力の書き込みに使用する 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ファイルにパッケージ化します。コードのコンパイルとパッケージ化には次の 2 通りの方法があります。

   • コマンドライン 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. [Bucket name (バケット名)] フィールドにka-app-code-<username>と入力します。バケット名にユーザー名などのサフィックスを追加して、グローバルに一意にします。[Next (次へ)] を選択します。

4. 設定オプションのステップでは、設定をそのままにし、[次へ] を選択します。

5. アクセス許可の設定のステップでは、設定をそのままにし、[次へ] を選択します。

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. [Create application] を選択します。

1. でIAMコンソールを開きますhttps://console.aws.amazon.com/iam/。

2. [Policies] (ポリシー) を選択します。前のセクションでコンソールによって作成された 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. [Configure application] ページで、[Code location] を次のように指定します。

   • [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. [Properties] の [Group ID] には、ProducerConfigPropertiesと入力します。

5. 次のアプリケーションのプロパティと値を入力します。

   グループ ID	キー	値
   ProducerConfigProperties	flink.inputstream.initpos	LATEST
   ProducerConfigProperties	aws.region	us-west-2
   ProducerConfigProperties	AggregationEnabled	false

6. [Monitoring] の [Monitoring metrics level] が [Application] に設定されていることを確認します。

7. CloudWatch をログに記録するには、有効化チェックボックスをオンにします。

8. [Update] (更新) を選択します。

IAM ロールを作成するには

1. でIAMコンソールを開きますhttps://console.aws.amazon.com/iam/。

2. ナビゲーションペインで [Roles (ロール)]、[Create Role (ロールの作成)] の順に選択します。

3. [信頼されるエンティティの種類を選択] で、[AWS のサービス] を選択します。[このロールを使用するサービスを選択] で、[Kinesis Analytics] を選択します。[ユースケースの選択] で、[Kinesis Analytics ] を選択します。

   [Next: Permissions] (次へ: アクセス許可) を選択します。

4. [アクセス権限ポリシーをアタッチする] ページで、[Next: Review] (次: 確認) を選択します。ロールを作成した後に、アクセス許可ポリシーをアタッチします。

5. [Create role (ロールの作成)] ページで、ロールの名前に MF-stream-rw-role を入力します。[ロールの作成] を選択します。

   これで、 という名前の新しいIAMロールが作成されましたMF-stream-rw-role。次に、ロールの信頼ポリシーとアクセス許可ポリシーを更新します。

6. アクセス許可ポリシーをロールにアタッチします。

   注記

   この演習では、Managed Service for Apache Flink が、Kinesis データストリーム (ソース) からのデータの読み取りと、別の Kinesis データストリームへの出力の書き込みの両方を実行するためにこのロールを引き受けます。このため、前のステップで作成したポリシー、アクセス許可ポリシーを作成する をアタッチします。

   1. [概要] ページで、[アクセス許可] タブを選択します。

   2. [Attach Policies (ポリシーのアタッチ)] を選択します。

   3. 検索ボックスにAKReadSourceStreamWriteSinkStream(前のセクションで作成したポリシー) と入力します。

   4. AKReadSourceStreamWriteSinkStream ポリシーを選択し、ポリシーのアタッチ を選択します。

1. 次のJSONコードを という名前のファイルに保存しますcreate_request.json。サンプルロールを、以前に作成したロールARNの ARNに置き換えます。バケットARNサフィックス (username) を前のセクションで選択したサフィックスに置き換えます。サービス実行ロールのサンプルのアカウント ID (012345678901) を、自分のアカウント ID に置き換えます。

   {
       "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 アカウント E メールアドレスを入力して、アカウント所有者AWS Management Consoleとして にサインインします。次のページでパスワードを入力します。

   ルートユーザーを使用してサインインする方法については、AWS サインイン ユーザーガイドのルートユーザーとしてサインインするを参照してください。

2. ルートユーザーの多要素認証 (MFA) を有効にします。

   手順については、「 ユーザーガイド」の AWS アカウント 「ルートユーザー (コンソール) の仮想MFAデバイスの有効化」を参照してください。 IAM

管理アクセスを持つユーザーを作成する

1. IAM Identity Center を有効にします。

   手順については、「AWS IAM Identity Center ユーザーガイド」の「AWS IAM Identity Centerの有効化」を参照してください。

2. IAM Identity Center で、ユーザーに管理アクセスを許可します。

   を ID ソース IAM アイデンティティセンターディレクトリ として使用する方法のチュートリアルについては、 AWS IAM Identity Center ユーザーガイドの「デフォルトを使用してユーザーアクセスを設定する IAM アイデンティティセンターディレクトリ」を参照してください。

管理アクセス権を持つユーザーとしてサインインする

• IAM Identity Center ユーザーでサインインするには、IAMIdentity Center ユーザーの作成時に E メールアドレスに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. 管理者ユーザーの名前付きプロファイルを config ファイルに追加します AWS CLI 。 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. コマンドプロンプトで以下のヘルプコマンドを入力して、セットアップを確認します。

   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. アプリケーションが出力の書き込みに使用する 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ファイルにパッケージ化します。コードのコンパイルとパッケージ化には次の 2 通りの方法があります。

   • コマンドライン 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. [Bucket name (バケット名)] フィールドにka-app-code-<username>と入力します。バケット名にユーザー名などのサフィックスを追加して、グローバルに一意にします。[Next (次へ)] を選択します。

4. 設定オプションのステップでは、設定をそのままにし、[次へ] を選択します。

5. アクセス許可の設定のステップでは、設定をそのままにし、[次へ] を選択します。

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. [Create application] を選択します。

1. でIAMコンソールを開きますhttps://console.aws.amazon.com/iam/。

2. [Policies] (ポリシー) を選択します。前のセクションでコンソールによって作成された 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. [Configure application] ページで、[Code location] を次のように指定します。

   • [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. [Monitoring] の [Monitoring metrics level] が [Application] に設定されていることを確認します。

6. CloudWatch をログに記録するには、有効化チェックボックスをオンにします。

7. [Update] (更新) を選択します。

1. MyApplication ページで、 の実行 を選択します。アクションを確認します。

2. アプリケーションが実行されたら、ページを更新します。コンソールには [Application graph] が示されます。

IAM ロールを作成するには

1. でIAMコンソールを開きますhttps://console.aws.amazon.com/iam/。

2. ナビゲーションペインで [Roles (ロール)]、[Create Role (ロールの作成)] の順に選択します。

3. [信頼されるエンティティの種類を選択] で、[AWS のサービス] を選択します。[このロールを使用するサービスを選択] で、[Kinesis Analytics] を選択します。[ユースケースの選択] で、[Kinesis Analytics ] を選択します。

   [Next: Permissions] (次へ: アクセス許可) を選択します。

4. [アクセス権限ポリシーをアタッチする] ページで、[Next: Review] (次: 確認) を選択します。ロールを作成した後に、アクセス許可ポリシーをアタッチします。

5. [Create role (ロールの作成)] ページで、ロールの名前に MF-stream-rw-role を入力します。[ロールの作成] を選択します。

   これで、 という名前の新しいIAMロールが作成されましたMF-stream-rw-role。次に、ロールの信頼ポリシーとアクセス許可ポリシーを更新します。

6. アクセス許可ポリシーをロールにアタッチします。

   注記

   この演習では、Managed Service for Apache Flink が、Kinesis データストリーム (ソース) からのデータの読み取りと、別の Kinesis データストリームへの出力の書き込みの両方を実行するためにこのロールを引き受けます。このため、前のステップで作成したポリシー、アクセス許可ポリシーを作成する をアタッチします。

   1. [概要] ページで、[アクセス許可] タブを選択します。

   2. [Attach Policies (ポリシーのアタッチ)] を選択します。

   3. 検索ボックスにAKReadSourceStreamWriteSinkStream(前のセクションで作成したポリシー) と入力します。

   4. AKReadSourceStreamWriteSinkStream ポリシーを選択し、ポリシーのアタッチ を選択します。

1. 次のJSONコードを という名前のファイルに保存しますcreate_request.json。サンプルロールを、以前に作成したロールARNの ARNに置き換えます。バケットARNサフィックス (username) を前のセクションで選択したサフィックスに置き換えます。サービス実行ロールのサンプルのアカウント ID (012345678901) を、自分のアカウント ID に置き換えます。

   {
       "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. Managed Service for Apache Flink コンソールを https://console.aws.amazon.com/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 アカウント E メールアドレスを入力して、アカウント所有者AWS Management Consoleとして にサインインします。次のページでパスワードを入力します。

   ルートユーザーを使用してサインインする方法については、AWS サインイン ユーザーガイドのルートユーザーとしてサインインするを参照してください。

2. ルートユーザーの多要素認証 (MFA) を有効にします。

   手順については、「 ユーザーガイド」の AWS アカウント 「ルートユーザー (コンソール) の仮想MFAデバイスの有効化」を参照してください。 IAM

管理アクセスを持つユーザーを作成する

1. IAM Identity Center を有効にします。

   手順については、「AWS IAM Identity Center ユーザーガイド」の「AWS IAM Identity Centerの有効化」を参照してください。

2. IAM Identity Center で、ユーザーに管理アクセスを許可します。

   を ID ソース IAM アイデンティティセンターディレクトリ として使用する方法のチュートリアルについては、 AWS IAM Identity Center ユーザーガイドの「デフォルトを使用してユーザーアクセスを設定する IAM アイデンティティセンターディレクトリ」を参照してください。

管理アクセス権を持つユーザーとしてサインインする

• IAM Identity Center ユーザーでサインインするには、IAMIdentity Center ユーザーの作成時に E メールアドレスに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. 管理者ユーザーの名前付きプロファイルを config ファイルに追加します AWS CLI 。 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. コマンドプロンプトで以下のヘルプコマンドを入力して、セットアップを確認します。

   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. アプリケーションが出力の書き込みに使用する 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ファイルにパッケージ化します。コードのコンパイルとパッケージ化には次の 2 通りの方法があります。

   • コマンドライン Maven ツールを使用します。JAR ファイルを含むディレクトリで次のコマンドを実行して、pom.xmlファイルを作成します。

     mvn package

     注記

     -DFlink.version パラメーターは、Apache Flink ランタイムのマネージドサービスバージョン 1.0.1 には必要ありません。バージョン 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. [Bucket name (バケット名)] フィールドにka-app-code-<username>と入力します。バケット名にユーザー名などのサフィックスを追加して、グローバルに一意にします。[Next (次へ)] を選択します。

4. 設定オプションのステップでは、設定をそのままにし、[次へ] を選択します。

5. アクセス許可の設定のステップでは、設定をそのままにし、[次へ] を選択します。

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] を選択します。

     注記

     Apache Flink 用 Managed Serviceは Apache Flink バージョン 1.8.2 または 1.6.2 を使用します。

   • バージョンプルダウンを「Apache Flink 1.6」に変更します。

4. アクセス許可 で、IAMロールの作成/更新 kinesis-analytics-MyApplication-us-west-2を選択します。

5. [Create application] を選択します。

1. でIAMコンソールを開きますhttps://console.aws.amazon.com/iam/。

2. [Policies] (ポリシー) を選択します。前のセクションでコンソールによって作成された 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. [Configure application] ページで、[Code location] を次のように指定します。

   • [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. [Monitoring] の [Monitoring metrics level] が [Application] に設定されていることを確認します。

6. CloudWatch をログに記録するには、有効化チェックボックスをオンにします。

7. [Update] (更新) を選択します。

1. MyApplication ページで、実行 を選択します。アクションを確認します。

2. アプリケーションが実行されたら、ページを更新します。コンソールには [Application graph] が示されます。

IAM ロールを作成するには

1. でIAMコンソールを開きますhttps://console.aws.amazon.com/iam/。

2. ナビゲーションペインで [Roles (ロール)]、[Create Role (ロールの作成)] の順に選択します。

3. [信頼されるエンティティの種類を選択] で、[AWS のサービス] を選択します。[このロールを使用するサービスを選択] で、[Kinesis Analytics] を選択します。[ユースケースの選択] で、[Kinesis Analytics ] を選択します。

   [Next: Permissions] (次へ: アクセス許可) を選択します。

4. [アクセス権限ポリシーをアタッチする] ページで、[Next: Review] (次: 確認) を選択します。ロールを作成した後に、アクセス許可ポリシーをアタッチします。

5. [Create role (ロールの作成)] ページで、ロールの名前に MF-stream-rw-role を入力します。[ロールの作成] を選択します。

   これで、 という名前の新しいIAMロールが作成されましたMF-stream-rw-role。次に、ロールの信頼ポリシーとアクセス許可ポリシーを更新します。

6. アクセス許可ポリシーをロールにアタッチします。

   注記

   この演習では、Managed Service for Apache Flink が、Kinesis データストリーム (ソース) からのデータの読み取りと、別の Kinesis データストリームへの出力の書き込みの両方を実行するためにこのロールを引き受けます。このため、前のステップで作成したポリシー、許可ポリシーを作成する をアタッチします。

   1. [概要] ページで、[アクセス許可] タブを選択します。

   2. [Attach Policies (ポリシーのアタッチ)] を選択します。

   3. 検索ボックスにAKReadSourceStreamWriteSinkStream(前のセクションで作成したポリシー) と入力します。

   4. AKReadSourceStreamWriteSinkStream ポリシーを選択し、ポリシーのアタッチ を選択します。

1. 次のJSONコードを という名前のファイルに保存しますcreate_request.json。サンプルロールを、以前に作成したロールARNの ARNに置き換えます。バケットARNサフィックス (username) を、前のセクションで選択したサフィックスに置き換えます。サービス実行ロールのサンプルのアカウント ID (012345678901) を、自分のアカウント ID に置き換えます。

   {
       "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] を選択します。

     注記

     Apache Flink 用 Managed Serviceは Apache Flink バージョン 1.15.2 を使用しています。

   • バージョンプルダウンは Apache Flink バージョン 1.15.2 (推奨バージョン) のままにしておきます。

4. アクセス許可 では、IAMロールの作成/更新 kinesis-analytics-MyApplication-us-west-2を選択します。

5. [Create application] を選択します。

1. でIAMコンソールを開きますhttps://console.aws.amazon.com/iam/。

2. [Policies] (ポリシー) を選択します。前のセクションでコンソールによって作成された 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. [Configure application] ページで、[Code location] を次のように指定します。

   • [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. [Monitoring] の [Monitoring metrics level] が [Application] に設定されていることを確認します。

5. CloudWatch をログに記録するには、有効化チェックボックスをオンにします。

6. [Update] (更新) を選択します。

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. [Create application] を選択します。

1. でIAMコンソールを開きますhttps://console.aws.amazon.com/iam/。

2. [Policies] (ポリシー) を選択します。前のセクションでコンソールによって作成された 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. [Configure application] ページで、[Code location] を次のように指定します。

   • [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. [Monitoring] の [Monitoring metrics level] が [Application] に設定されていることを確認します。

5. CloudWatch をログに記録するには、有効化チェックボックスをオンにします。

6. [Update] (更新) を選択します。

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. [Create application] を選択します。

   注記

   コンソールを使用して Managed Service for Apache Flink アプリケーションを作成する場合、アプリケーション用にIAMロールとポリシーを作成するオプションがあります。アプリケーションではこのロールとポリシーを使用して、依存リソースにアクセスします。これらのIAMリソースには、アプリケーション名とリージョンを使用して次のように名前が付けられます。

   • [アプリケーション名] にはMyApplicationと入力します。

   • [ランタイム] には、[Apache Flink] を選択します。

   • バージョンはApache Flink バージョン 1.15.2 (推薦バージョン)のままにしておきます。

6. アクセス許可 で、IAMロール の作成/更新 kinesis-analytics-MyApplication-us-west-2を選択します。

7. [Create application] を選択します。

1. でIAMコンソールを開きますhttps://console.aws.amazon.com/iam/。

2. [Policies] (ポリシー) を選択します。前のセクションでコンソールによって作成された 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. [Configure application] ページで、[Code location] を次のように指定します。

   • [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. [Monitoring] の [Monitoring metrics level] が [Application] に設定されていることを確認します。

5. CloudWatch ログ記録するには、有効化チェックボックスをオンにします。

6. [Update] (更新) を選択します。

1. MyApplication ページで、実行 を選択します。「スナップショットなしで実行」オプションを選択したままにして、確定します。

2. アプリケーションが実行されたら、ページを更新します。コンソールには [Application graph] が示されます。

1. 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. 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. Managed Service for Apache Flink コンソールを https://console.aws.amazon.com/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. [Create application] を選択します。

1. MyApplication ページで、「 の設定」を選択します。

2. [Configure application] ページで、[Code location] を次のように指定します。

   • [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	1,000

   アプリケーションコードは、上記のアプリケーションプロパティを読み取り、 と Amazon MSKクラスターとの対話に使用されるソースVPCとシンクを設定します。プロパティ使用の詳細については、「Managed Service for Apache Flink でランタイムプロパティを使用する」を参照してください。

5. スナップショットで 無効 を選択します。これにより、無効なアプリケーション状態データを読み込まずにアプリケーションを簡単に更新できます。

6. [Monitoring] の [Monitoring metrics level] が [Application] に設定されていることを確認します。

7. CloudWatch をログに記録するには、有効化チェックボックスをオンにします。

8. Virtual Private Cloud (VPC） セクションVPCで、アプリケーションに関連付ける を選択します。アプリケーションがVPCリソースへのアクセスVPCに使用する に関連付けられたサブネットとセキュリティグループを選択します。

9. [Update] (更新) を選択します。

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. Managed Service for Apache Flink コンソールを https://console.aws.amazon.com/flink で開く

2. Managed Service for Apache Flinkのダッシュボードで、「分析アプリケーションの作成」 を選択します。

3. 「Managed Service for Apache Flink-アプリケーションの作成」ページで、次のようにアプリケーションの詳細を入力します。

   • [アプリケーション名] には MyApplication と入力します。

   • [ランタイム] には、[Apache Flink] を選択します。

     注記

     Apache Flink 用 Managed Serviceは Apache Flink バージョン 1.15.2 を使用しています。

   • バージョンプルダウンは Apache Flink バージョン 1.15.2 (推奨バージョン) のままにしておきます。

4. アクセス許可 については、IAMロールの作成/更新 kinesis-analytics-MyApplication-us-west-2を選択します。

5. [Create application] を選択します。

1. でIAMコンソールを開きますhttps://console.aws.amazon.com/iam/。

2. [Policies] (ポリシー) を選択します。前のセクションでコンソールによって作成された 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. [Configure application] ページで、[Code location] を次のように指定します。

   • [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. [Monitoring] の [Monitoring metrics level] が [Application] に設定されていることを確認します。

9. CloudWatch をログに記録するには、有効化チェックボックスをオンにします。

10. [Update] (更新) を選択します。

1. Managed Service for Apache Flink コンソールを https://console.aws.amazon.com/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 バージョンで Apache Flink Kinesis Streams コネクタを使用する」を参照してください。

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. Managed Service for Apache Flink コンソールを https://console.aws.amazon.com/flink で開く

2. Managed Service for Apache Flinkのダッシュボードで、「分析アプリケーションの作成」 を選択します。

3. 「Managed Service for Apache Flink-アプリケーションの作成」ページで、次のようにアプリケーションの詳細を入力します。

   • [アプリケーション名] には MyApplication と入力します。

   • [Description (説明)] に My java test app と入力します。

   • [ランタイム] には、[Apache Flink] を選択します。

     注記

     Apache Flink 用 Managed Serviceは Apache Flink バージョン 1.15.2 を使用しています。

   • バージョンプルダウンは Apache Flink バージョン 1.15.2 (推奨バージョン) のままにしておきます。

4. アクセス許可 で、IAMロール の作成/更新 kinesis-analytics-MyApplication-us-west-2を選択します。

5. [Create application] を選択します。

1. でIAMコンソールを開きますhttps://console.aws.amazon.com/iam/。

2. [Policies] (ポリシー) を選択します。前のセクションでコンソールによって作成された 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. [Configure application] ページで、[Code location] を次のように指定します。

   • [Amazon S3 バケット] で、ka-app-code-<username>と入力します。

   • [Amazon S3 オブジェクトへのパス] で、java-getting-started-1.0.jarと入力します。

3. アプリケーションリソースへのアクセス のアクセス許可 で、IAMロールの作成/更新 kinesis-analytics-MyApplication-us-west-2を選択します。

4. [Monitoring] の [Monitoring metrics level] が [Application] に設定されていることを確認します。

5. CloudWatch ログ記録するには、有効化チェックボックスをオンにします。

6. [Update] (更新) を選択します。

IAM ロールを作成するには

1. でIAMコンソールを開きますhttps://console.aws.amazon.com/iam/。

2. ナビゲーションペインで [Roles (ロール)]、[Create Role (ロールの作成)] の順に選択します。

3. [信頼されるエンティティの種類を選択] で、[AWS のサービス] を選択します。[このロールを使用するサービスを選択] で、[Kinesis Analytics] を選択します。[ユースケースの選択] で、[Kinesis Analytics ] を選択します。

   [Next: Permissions] (次へ: アクセス許可) を選択します。

4. [アクセス権限ポリシーをアタッチする] ページで、[Next: Review] (次: 確認) を選択します。ロールを作成した後に、アクセス許可ポリシーをアタッチします。

5. [Create role (ロールの作成)] ページで、ロールの名前に MF-stream-rw-role を入力します。[ロールの作成] を選択します。

   これで、 という名前の新しいIAMロールが作成されましたMF-stream-rw-role。次に、ロールの信頼ポリシーとアクセス許可ポリシーを更新します。

6. アクセス許可ポリシーをロールにアタッチします。

   注記

   この演習では、Managed Service for Apache Flink が、Kinesis データストリーム (ソース) からのデータの読み取りと、別の Kinesis データストリームへの出力の書き込みの両方を実行するためにこのロールを引き受けます。このため、前のステップで作成したポリシー、許可ポリシーを作成する をアタッチします。

   1. [概要] ページで、[アクセス許可] タブを選択します。

   2. [Attach Policies (ポリシーのアタッチ)] を選択します。

   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. [ロググループの削除]を選択し、削除を確認してください。

1. 入門チュートリアルからkinesis-analytics-service-MyApplication-us-west-2ポリシーを編集します。このポリシーにより、ソース ストリームを読み取るためにソースアカウントのロールを引き受けることができます。

   注記

   コンソールを使用してアプリケーションを作成する場合、コンソールはkinesis-analytics-service-<application name>-<application region>というポリシーとkinesisanalytics-<application name>-<application region>というロールを作成します。

   以下の強調表示されたセクションをポリシーに追加します。サンプルアカウント ID (SOURCE01234567) は、ソースストリームに使用するアカウントの ID を使用します。

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "AssumeRoleInSourceAccount",
               "Effect": "Allow",
               "Action": "sts:AssumeRole",
               "Resource": "arn:aws:iam::SOURCE01234567:role/KA-Source-Stream-Role"
           },
           {
               "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": "ListCloudwatchLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:SINK012345678:log-group:*"
               ]
           },
           {
               "Sid": "ListCloudwatchLogStreams",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogStreams"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:SINK012345678:log-group:/aws/kinesis-analytics/MyApplication:log-stream:*"
               ]
           },
           {
               "Sid": "PutCloudwatchLogs",
               "Effect": "Allow",
               "Action": [
                   "logs:PutLogEvents"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:SINK012345678:log-group:/aws/kinesis-analytics/MyApplication:log-stream:kinesis-analytics-log-stream"
               ]
           }
       ]
   }
   

2. kinesis-analytics-MyApplication-us-west-2 ロールを開き、Amazon リソースネーム () を書き留めますARN。これは次のセクションで必要になります。ロールARNは次のようになります。

   arn:aws:iam::SINK012345678:role/service-role/kinesis-analytics-MyApplication-us-west-2