Apache Beam を使用してアプリケーションを作成する

次の内容で、ping.py という名前のファイルを作成します。

import json
import boto3
import random

kinesis = boto3.client('kinesis')

while True:
        data = random.choice(['ping', 'telnet', 'ftp', 'tracert', 'netstat'])
        print(data)
        kinesis.put_record(
                StreamName="ExampleInputStream",
                Data=data,
                PartitionKey="partitionkey")

ping.py スクリプトを実行します。

$ python ping.py

チュートリアルの残りの部分を完了する間、スクリプトを実行し続けてください。

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

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

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

amazon-kinesis-data-analytics-java-examples/Beam ディレクトリに移動します。

Java と Maven がまだインストールされていない場合は、インストールします。詳細については、チュートリアル: Managed Service for Apache Flink で の使用 DataStream APIを開始するチュートリアルの「必要な前提条件を完了する」を参照してください。

次のコマンドを使用して、アプリケーションをコンパイルします。

mvn package -Dflink.version=1.15.2 -Dflink.version.minor=1.8

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

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

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

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

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

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

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

[Create application] を選択します。

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

[Policies] (ポリシー) を選択します。前のセクションでコンソールによって作成された kinesis-analytics-service-MyApplication-us-west-2 ポリシーを選択します。

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

次のポリシー例で強調表示されているセクションをポリシーに追加します。サンプルアカウント IDs (012345678901) を アカウント ID で使用します。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "ReadCode",
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "logs:DescribeLogGroups",
                "s3:GetObjectVersion"
            ],
            "Resource": [
                "arn:aws:logs:us-west-2:012345678901:log-group:*",
                "arn:aws:s3:::ka-app-code-<username>/basic-beam-app-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"
        }
    ]
}

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

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

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

次のように入力します。

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

kinesis-analytics-MyApplication-us-west-2 ロールを選択します。

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

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

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

/aws/kinesis-analytics/MyApplication ロググループを選択します。

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