Managed Service for Apache Flink の以前のバージョン情報

1. 「Apache Maven」がインストールされていて、 JAVA_HOME 環境変数が JRE ではなく JDK を指していることを確認してください。次のコマンドで 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/signup を開きます。

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

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

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

を保護する AWS アカウントのルートユーザー

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

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

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

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

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

1. IAM アイデンティティセンターを有効にします。

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

2. IAM アイデンティティセンターで、ユーザーに管理アクセスを付与します。

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

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

• IAM アイデンティティセンターのユーザーとしてサインインするには、IAM アイデンティティセンターのユーザーの作成時に E メールアドレスに送信されたサインイン URL を使用します。

  IAM Identity Center ユーザーを使用してサインインする方法については、「 ユーザーガイド」の AWS 「 アクセスポータルにサインインする」を参照してください。 AWS サインイン

追加のユーザーにアクセス権を割り当てる

1. IAM アイデンティティセンターで、最小特権のアクセス許可を適用するというベストプラクティスに従ったアクセス許可セットを作成します。

   手順については、「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 ツールを使用します。pom.xml ファイルが格納されているディレクトリで次のコマンドを実行して JAR ファイルを作成します。

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

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. [ポリシー] を選択します。前のセクションでコンソールによって作成された kinesis-analytics-service-MyApplication-us-west-2 ポリシーを選択します。

3. [概要] ページで、[ポリシーの編集] を選択します。[JSON] タブを選択します。

4. 次のポリシー例で強調表示されているセクションをポリシーに追加します。サンプルのアカウント ID (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] ページで、[Congirue] を選択します。

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

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

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

3. [Access to application resources] の [Access permissions] では、[Create / update IAM rolekinesis-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 logging] では、[Enable] チェックボックスをオンにします。

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 を入力します。[ロールの作成] を選択します。

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

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. Kinesis コンソール (https://console.aws.amazon.com/kinesis) を開きます。

2. Apache Flink 用 Managed Serviceパネルで、MyApplication を選択します。

3. アプリケーションのページで[削除]を選択し、削除を確認します。

1. https://console.aws.amazon.com/flink で Apache Flink 用 Managed Serviceコンソールを開く

2. Kinesis Data Streams パネルで、「ExampleInputStream」を選択します。

3. 「ExampleInputStream」ページで、「Kinesis ストリームを削除」を選択し、削除を確定します。

4. 「Kinesis ストリーム」ページで、「ExampleOutputStream」を選択し、「アクション」を選択し、「削除」を選択し、削除を確定します。

1. https://console.aws.amazon.com/s3/ で Amazon 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/signup を開きます。

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

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

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

を保護する AWS アカウントのルートユーザー

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

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

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

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

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

1. IAM アイデンティティセンターを有効にします。

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

2. IAM アイデンティティセンターで、ユーザーに管理アクセスを付与します。

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

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

• IAM アイデンティティセンターのユーザーとしてサインインするには、IAM アイデンティティセンターのユーザーの作成時に E メールアドレスに送信されたサインイン URL を使用します。

  IAM Identity Center ユーザーを使用してサインインする方法については、「 ユーザーガイド」の AWS 「 アクセスポータルにサインインする」を参照してください。 AWS サインイン

追加のユーザーにアクセス権を割り当てる

1. IAM アイデンティティセンターで、最小特権のアクセス許可を適用するというベストプラクティスに従ったアクセス許可セットを作成します。

   手順については、「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 ツールを使用します。pom.xml ファイルが格納されているディレクトリで次のコマンドを実行して JAR ファイルを作成します。

     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 で Apache Flink 用 Managed Serviceコンソールを開く

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. [ポリシー] を選択します。前のセクションでコンソールによって作成された kinesis-analytics-service-MyApplication-us-west-2 ポリシーを選択します。

3. [概要] ページで、[ポリシーの編集] を選択します。[JSON] タブを選択します。

4. 次のポリシー例で強調表示されているセクションをポリシーに追加します。サンプルのアカウント ID (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] ページで、[Congirue] を選択します。

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

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

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

3. [Access to application resources] の [Access permissions] では、[Create / update IAM rolekinesis-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 logging] では、[Enable] チェックボックスをオンにします。

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 を入力します。[ロールの作成] を選択します。

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

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. Kinesis コンソール (https://console.aws.amazon.com/kinesis) を開きます。

2. Apache Flink 用 Managed Serviceパネルで、MyApplication を選択します。

3. アプリケーションのページで[削除]を選択し、削除を確認します。

1. https://console.aws.amazon.com/flink で Apache Flink 用 Managed Serviceコンソールを開く

2. Kinesis Data Streams パネルで、「ExampleInputStream」を選択します。

3. 「ExampleInputStream」ページで、「Kinesis ストリームを削除」を選択し、削除を確定します。

4. 「Kinesis ストリーム」ページで、「ExampleOutputStream」を選択し、「アクション」を選択し、「削除」を選択し、削除を確定します。

1. https://console.aws.amazon.com/s3/ で Amazon 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/signup を開きます。

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

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

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

を保護する AWS アカウントのルートユーザー

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

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

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

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

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

1. IAM アイデンティティセンターを有効にします。

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

2. IAM アイデンティティセンターで、ユーザーに管理アクセスを付与します。

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

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

• IAM アイデンティティセンターのユーザーとしてサインインするには、IAM アイデンティティセンターのユーザーの作成時に E メールアドレスに送信されたサインイン URL を使用します。

  IAM Identity Center ユーザーを使用してサインインする方法については、「 ユーザーガイド」の AWS 「 アクセスポータルにサインインする」を参照してください。 AWS サインイン

追加のユーザーにアクセス権を割り当てる

1. IAM アイデンティティセンターで、最小特権のアクセス許可を適用するというベストプラクティスに従ったアクセス許可セットを作成します。

   手順については、「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 ツールを使用します。pom.xml ファイルが格納されているディレクトリで次のコマンドを実行して JAR ファイルを作成します。

     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 で Apache Flink 用 Managed Serviceコンソールを開く

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. [ポリシー] を選択します。前のセクションでコンソールによって作成された kinesis-analytics-service-MyApplication-us-west-2 ポリシーを選択します。

3. [概要] ページで、[ポリシーの編集] を選択します。[JSON] タブを選択します。

4. 次のポリシー例で強調表示されているセクションをポリシーに追加します。サンプルのアカウント ID (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] ページで、[Congirue] を選択します。

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

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

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

3. [Access to application resources] の [Access permissions] では、[Create / update IAM rolekinesis-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 logging] では、[Enable] チェックボックスをオンにします。

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

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

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 を入力します。[ロールの作成] を選択します。

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

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. Kinesis コンソール (https://console.aws.amazon.com/kinesis) を開きます。

2. Apache Flink 用 Managed Serviceパネルで、MyApplication を選択します。

3. [設定] を選択します。

4. スナップショットセクションで「無効」を選択して、更新を選択します。

5. アプリケーションのページで[削除]を選択し、削除を確認します。

1. https://console.aws.amazon.com/flink で Apache Flink 用 Managed Serviceコンソールを開く

2. Kinesis Data Streams パネルで、「ExampleInputStream」を選択します。

3. 「ExampleInputStream」ページで、「Kinesis ストリームを削除」を選択し、削除を確定します。

4. 「Kinesis ストリーム」ページで、「ExampleOutputStream」を選択し、「アクション」を選択し、「削除」を選択し、削除を確定します。

1. https://console.aws.amazon.com/s3/ で Amazon 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/signup を開きます。

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

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

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

を保護する AWS アカウントのルートユーザー

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

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

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

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

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

1. IAM アイデンティティセンターを有効にします。

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

2. IAM アイデンティティセンターで、ユーザーに管理アクセスを付与します。

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

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

• IAM アイデンティティセンターのユーザーとしてサインインするには、IAM アイデンティティセンターのユーザーの作成時に E メールアドレスに送信されたサインイン URL を使用します。

  IAM Identity Center ユーザーを使用してサインインする方法については、「 AWS サインイン ユーザーガイド」の AWS 「 アクセスポータルにサインインする」を参照してください。

追加のユーザーにアクセス権を割り当てる

1. IAM アイデンティティセンターで、最小特権のアクセス許可を適用するというベストプラクティスに従ったアクセス許可セットを作成します。

   手順については、「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 ツールを使用します。pom.xml ファイルが格納されているディレクトリで次のコマンドを実行して JAR ファイルを作成します。

     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 で Apache Flink 用 Managed Serviceコンソールを開く

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. [ポリシー] を選択します。前のセクションでコンソールによって作成された kinesis-analytics-service-MyApplication-us-west-2 ポリシーを選択します。

3. [概要] ページで、[ポリシーの編集] を選択します。[JSON] タブを選択します。

4. 次のポリシー例で強調表示されているセクションをポリシーに追加します。サンプルのアカウント ID (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] ページで、[Congirue] を選択します。

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

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

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

3. [Access to application resources] の [Access permissions] では、[Create / update IAM rolekinesis-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 logging] では、[Enable] チェックボックスをオンにします。

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

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

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 を入力します。[ロールの作成] を選択します。

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

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. Kinesis コンソール (https://console.aws.amazon.com/kinesis) を開きます。

2. Apache Flink 用 Managed Serviceパネルで、MyApplication を選択します。

3. [設定] を選択します。

4. スナップショットセクションで「無効」を選択して、更新を選択します。

5. アプリケーションのページで[削除]を選択し、削除を確認します。

1. https://console.aws.amazon.com/flink で Apache Flink 用 Managed Serviceコンソールを開く

2. Kinesis Data Streams パネルで、「ExampleInputStream」を選択します。

3. 「ExampleInputStream」ページで、「Kinesis ストリームを削除」を選択し、削除を確定します。

4. 「Kinesis ストリーム」ページで、「ExampleOutputStream」を選択し、「アクション」を選択し、「削除」を選択し、削除を確定します。

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

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. [ポリシー] を選択します。前のセクションでコンソールによって作成された kinesis-analytics-service-MyApplication-us-west-2 ポリシーを選択します。

3. [概要] ページで、[ポリシーの編集] を選択します。[JSON] タブを選択します。

4. 次のポリシー例で強調表示されているセクションをポリシーに追加します。サンプルのアカウント ID (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] ページで、[Congirue] を選択します。

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

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

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

3. [Access to application resources] の [Access permissions] では、[Create / update IAM rolekinesis-analytics-MyApplication-us-west-2] を選択します。

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

5. [CloudWatch logging] では、[Enable] チェックボックスをオンにします。

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

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

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

1. https://console.aws.amazon.com/flink で Apache Flink 用 Managed Serviceコンソールを開く

2. Apache Flink 用 Managed Serviceパネルで、MyApplication を選択します。

3. アプリケーションのページで[削除]を選択し、削除を確認します。

1. 「https://console.aws.amazon.com/kinesis」で Kinesis コンソールを開きます。

2. Kinesis Data Streams パネルで、「ExampleInputStream」を選択します。

3. 「ExampleInputStream」ページで、「Kinesis ストリームを削除」を選択し、削除を確定します。

4. 「Kinesis ストリーム」ページで、「ExampleOutputStream」を選択し、「アクション」を選択し、「削除」を選択し、削除を確定します。

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

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. [ポリシー] を選択します。前のセクションでコンソールによって作成された kinesis-analytics-service-MyApplication-us-west-2 ポリシーを選択します。

3. [概要] ページで、[ポリシーの編集] を選択します。[JSON] タブを選択します。

4. 次のポリシー例で強調表示されているセクションをポリシーに追加します。サンプルのアカウント ID (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] ページで、[Congirue] を選択します。

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

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

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

3. [Access to application resources] の [Access permissions] では、[Create / update IAM rolekinesis-analytics-MyApplication-us-west-2] を選択します。

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

5. [CloudWatch logging] では、[Enable] チェックボックスをオンにします。

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

1. https://console.aws.amazon.com/flink で Apache Flink 用 Managed Serviceコンソールを開く

2. Apache Flink 用 Managed Serviceパネルで、MyApplication を選択します。

3. アプリケーションのページで[削除]を選択し、削除を確認します。

1. 「https://console.aws.amazon.com/kinesis」で Kinesis コンソールを開きます。

2. Kinesis Data Streams パネルで、「ExampleInputStream」を選択します。

3. 「ExampleInputStream」ページで、「Kinesis ストリームを削除」を選択し、削除を確定します。

4. 「Kinesis ストリーム」ページで、「ExampleOutputStream」を選択し、「アクション」を選択し、「削除」を選択し、削除を確定します。

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

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

   注記

   コンソールを使用して Apache Flink アプリケーション用 Managed Service を作成する場合は、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. [ポリシー] を選択します。前のセクションでコンソールによって作成された kinesis-analytics-service-MyApplication-us-west-2 ポリシーを選択します。

3. [概要] ページで、[ポリシーの編集] を選択します。[JSON] タブを選択します。

4. 次のポリシー例で強調表示されているセクションをポリシーに追加します。サンプルのアカウント ID (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] ページで、[Congirue] を選択します。

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

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

   • [Amazon S3 オブジェクトへのパス] で、code/aws-kinesis-analytics-java-apps-1.0.jarと入力します。

3. [Access to application resources] の [Access permissions] では、[Create / update IAM rolekinesis-analytics-MyApplication-us-west-2] を選択します。

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

5. [CloudWatch logging] では、[Enable] チェックボックスをオンにします。

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

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

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 で Apache Flink 用 Managed Serviceコンソールを開く

2. Apache Flink 用 Managed Serviceパネルで、MyApplication を選択します。

3. アプリケーションページで[削除]を選択し、削除を確定します。

1. 「https://console.aws.amazon.com/kinesis」で Kinesis コンソールを開きます。

2. Kinesis Data Streams パネルで、「ExampleInputStream」を選択します。

3. 「ExampleInputStream」ページで、「Kinesis ストリームを削除」を選択し、削除を確定します。

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

2. ka-app-code-<username＞バケットを選択します。

3. [削除] を選択し、バケット名を入力して削除を確認します。

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

2. ナビゲーションバーで、[ポリシー] を選択します。

3. フィルターコントロールに「kinesis」と入力します。

4. 「kinesis-analytics-service-MyApplication-us-west-2」ポリシーを選択します。

5. [ポリシーアクション]、[削除] の順に選択します。

6. ナビゲーションバーで ロールを選択します。

7. Kinesis-Analytics-MyApplication-us-west-2ロールを選択します。

8. [ロールの削除] を選択し、削除を確定します。

1. CloudWatch コンソール (https://console.aws.amazon.com/cloudwatch/) を開きます。

2. ナビゲーションバーで [ログ] を選択します。

3. 「/aws/kinesis-analytics/MyApplication」ロググループを選択してください。

4. [ロググループの削除]を選択し、削除を確認してください。

1. Git クライアントをまだインストールしていない場合は、インストールします。詳細については、「Git のインストール」をご参照ください。

2. 次のコマンドを使用してリモートリポジトリのクローンを作成します。

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

3. アプリケーションコードはamazon-kinesis-data-analytics-java-examples/KafkaConnectors/KafkaGettingStartedJob.javaファイルに含まれています。コードを調べて、Managed Service for Apache Flink アプリケーションコードの構造を了解することができます。

4. コマンドライン Maven ツールまたは優先的な開発環境を使用して JAR ファイルを作成します。コマンドライン Maven ツールを使用して JAR ファイルをコンパイルするには、次のように入力します。

   mvn package -Dflink.version=1.15.3

   ビルドが成功する場合、次のファイルが作成されます。

   target/KafkaGettingStartedJob-1.0.jar

   注記

   提供されているソースコードは Java 11 のライブラリーに依存しています。開発環境を使用している場合は、

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

2. ファイルの選択のステップで、[ファイルを追加] を選択します。前のステップで作成した KafkaGettingStartedJob-1.0.jar ファイルに移動します。

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

1. https://console.aws.amazon.com/flink. で Managed Service for Apache Flink コンソールを開きます。

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

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

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

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

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

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

1. [MyApplication] ページで、[Congirue] を選択します。

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

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

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

3. [Access to application resources] の [Access permissions] では、[Create / update IAM rolekinesis-analytics-MyApplication-us-west-2] を選択します。

   注記

   コンソールを使用してアプリケーションリソース (CloudWatch Logs や Amazon VPC など) を指定すると、コンソールはアプリケーション実行ロールを変更して、それらのリソースにアクセスする権限を付与します。

4. [プロパティ] で [グループの追加]を選択します。以下のプロパティを入力します。

   グループ ID	キー	値
   KafkaSource	トピック	AWS KafkaTutorialTopic
   KafkaSource	bootstrap.servers	以前に保存したブートストラップサーバーリスト
   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	以前に保存したブートストラップサーバーリスト
   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

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

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

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

7. [CloudWatch logging] では、[有効化] チェックボックスをオンにします。

8. [仮想プライベートクラウド (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. https://console.aws.amazon.com/flink で Apache Flink 用 Managed Serviceコンソールを開く

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. [ポリシー] を選択します。前のセクションでコンソールによって作成された kinesis-analytics-service-MyApplication-us-west-2 ポリシーを選択します。

3. [概要] ページで、[ポリシーの編集] を選択します。[JSON] タブを選択します。

4. 次のポリシー例で強調表示されているセクションをポリシーに追加します。サンプルのアカウント ID (012345678901) を自分のアカウント ID に置き換えます。

   注記

   これらの権限により、アプリケーションには EFO Consumerへのアクセス権限が付与されます。

   {
       "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] ページで、[Congirue] を選択します。

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

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

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

3. [Access to application resources] の [Access permissions] では、[Create / update IAM rolekinesis-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 logging] では、[Enable] チェックボックスをオンにします。

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

1. https://console.aws.amazon.com/flink で Apache Flink 用 Managed Serviceコンソールを開く

2. Apache Flink 用 Managed Serviceパネルで、MyApplication を選択します。

3. アプリケーションのページで[削除]を選択し、削除を確認します。

1. 「https://console.aws.amazon.com/kinesis」で Kinesis コンソールを開きます。

2. Kinesis Data Streams パネルで、「ExampleInputStream」を選択します。

3. 「ExampleInputStream」ページで、「Kinesis ストリームを削除」を選択し、削除を確定します。

4. 「Kinesis ストリーム」ページで、「ExampleOutputStream」を選択し、「アクション」を選択し、「削除」を選択し、削除を確定します。

1. https://console.aws.amazon.com/s3/ で Amazon 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. https://console.aws.amazon.com/s3/ で Amazon S3 コンソールを開きます。

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

3. ファイルの選択のステップで、[ファイルを追加] を選択します。前のステップで作成した java-getting-started-1.0.jar ファイルに移動します。

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

1. https://console.aws.amazon.com/flink で Apache Flink 用 Managed Serviceコンソールを開く

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. [ポリシー] を選択します。前のセクションでコンソールによって作成された kinesis-analytics-service-MyApplication-us-west-2 ポリシーを選択します。

3. [概要] ページで、[ポリシーの編集] を選択します。[JSON] タブを選択します。

4. 次のポリシー例で強調表示されているセクションをポリシーに追加します。サンプルのアカウント ID (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] ページで、[Congirue] を選択します。

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

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

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

3. [Access to application resources] の [Access permissions] では、[Create / update IAM rolekinesis-analytics-MyApplication-us-west-2] を選択します。

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

5. [CloudWatch logging] では、[Enable] チェックボックスをオンにします。

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 を入力します。[ロールの作成] を選択します。

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

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 で Apache Flink 用 Managed Serviceコンソールを開く

2. Apache Flink 用 Managed Serviceパネルで、MyApplication を選択します。

3. [設定] を選択します。

4. スナップショットセクションで「無効」を選択して、更新を選択します。

5. アプリケーションのページで[削除]を選択し、削除を確認します。

1. 「https://console.aws.amazon.com/kinesis」で Kinesis コンソールを開きます。

2. Kinesis Data Streams パネルで、「ExampleInputStream」を選択します。

3. 「ExampleInputStream」ページで、「Kinesis ストリームを削除」を選択し、削除を確定します。

1. Kinesis コンソール (https://console.aws.amazon.com/kinesis) を開きます。

2. Firehose パネルで ExampleDeliveryStream を選択します。

3. ExampleDeliveryStream ページで、Firehose ストリームの削除を選択し、削除を確定します。

1. https://console.aws.amazon.com/s3/ で Amazon 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. サービスロールの Amazon リソースネーム (ARN) をメモしておきます。これは次のセクションで必要になります。[the IAM role ARN number (IAM ロールの ARN 番号)] は、次のようになります。

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