Managed Service for Apache Flink アプリケーションを作成して実行する

データストリームを作成するには (AWS CLI)

1. 最初のストリーム (ExampleInputStream) を作成するには、次の Amazon Kinesis create-stream AWS CLI コマンドを使用します。

   $ aws kinesis create-stream \
   --stream-name ExampleInputStream \
   --shard-count 1 \
   --region us-east-1 \

2. アプリケーションが出力の書き込みに使用する 2 番目のストリームを作成するには、同じコマンドを実行し、ストリーム名を に変更しますExampleOutputStream。

   $ aws kinesis create-stream \
   --stream-name ExampleOutputStream \
   --shard-count 1 \
   --region us-east-1 \

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

   git clone https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples.git

2. amazon-managed-service-for-apache-flink-examples/tree/main/java/GettingStarted ディレクトリに移動します。

レコードを観察するには

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

2. このチュートリアルを実行しているリージョンが同じであることを確認します。デフォルトでは us-east-1 US East (バージニア北部) です。一致しない場合は、リージョンを変更します。

3. データストリーム を選択します。

4. 監視するストリームを ExampleInputStreamまたは のいずれかで選択します。 ExampleOutputStream.

5. データビューワータブを選択します。

6. 任意のシャード を選択し、最新の を開始位置 として保持し、レコードの取得 を選択します。このリクエストの「レコードが見つかりません」エラーが表示される場合があります。その場合は、レコードの取得を再試行する を選択します。ストリームディスプレイに発行される最新のレコード。

7. データ列の値を選択して、レコードの内容を JSON 形式で検査します。

アプリケーションコードJARファイルをアップロードするには

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

2. アプリケーションコード用に以前に作成したバケットを選択します。

3. アップロードを選択します。

4. ファイルの追加を選択します。

5. 前のステップで生成されたJARファイルに移動します。 target/amazon-msf-java-stream-app-1.0.jar

6. 他の設定を変更せずにアップロードを選択します。

アプリケーションを作成するには

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

2. 正しいリージョンが選択されていることを確認します: us-east-1 US East (バージニア北部）

3. 右側のメニューを開き、Apache Flink アプリケーションを選択し、ストリーミングアプリケーションを作成します。または、最初のページのスタートコンテナでストリーミングアプリケーションを作成するを選択します。

4. ストリーミングアプリケーションの作成ページで、次の操作を行います。

   • ストリーム処理アプリケーションを設定する方法を選択します。「最初から作成」を選択します。

   • Apache Flink 設定、Application Flink バージョン： Apache Flink 1.19 を選択します。

5. アプリケーションを設定する

   • アプリケーション名： を入力しますMyApplication。

   • 説明： と入力しますMy java test app。

   • アプリケーションリソースへのアクセス： 必要なポリシーkinesis-analytics-MyApplication-us-east-1を使用してIAMロールを作成/更新する を選択します。

6. アプリケーション設定用のテンプレートを設定する

   • テンプレート： 開発 を選択します。

7. ページの下部にあるストリーミングアプリケーションの作成を選択します。

ポリシーを編集するには

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

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

3. 編集 を選択し、 JSONタブを選択します。

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

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:s3:::my-bucket/kinesis-analytics-placeholder-s3-object"
               ]
           },
           {
               "Sid": "ListCloudwatchLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:us-east-1:012345678901:log-group:*"
               ]
           },
           {
               "Sid": "ListCloudwatchLogStreams",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogStreams"
               ],
               "Resource": [
                   "arn:aws:logs:us-east-1:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:*"
               ]
           },
           {
               "Sid": "PutCloudwatchLogs",
               "Effect": "Allow",
               "Action": [
                   "logs:PutLogEvents"
               ],
               "Resource": [
                   "arn:aws:logs:us-east-1:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:kinesis-analytics-log-stream"
               ]
           },
           {
               "Sid": "ReadInputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-east-1:012345678901:stream/ExampleInputStream"
           },
           {
               "Sid": "WriteOutputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-east-1:012345678901:stream/ExampleOutputStream"
           }
       ]
   }

5. ページの下部にある次へ を選択し、変更の保存 を選択します。

設定を編集するには

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

2. アプリケーションコードの場所セクションで、次の操作を行います。

   • Amazon S3 バケット では、アプリケーションコード用に以前に作成したバケットを選択します。ブラウズを選択して正しいバケットを選択し、 を選択します。バケット名をクリックしないでください。

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

3. アクセス許可 では、必要なポリシー kinesis-analytics-MyApplication-us-east-1を使用してIAMロールを作成/更新する を選択します。

4. ランタイムプロパティセクションで、次のプロパティを追加します。

5. 新しい項目を追加 を選択し、次の各パラメータを追加します。

   グループ ID	キー	値
   InputStream0	stream.name	ExampleInputStream
   InputStream0	aws.region	us-east-1
   OutputStream0	stream.name	ExampleOutputStream
   OutputStream0	aws.region	us-east-1

6. 他のセクションは変更しないでください。

7. [Save changes] (変更の保存) をクリックします。

アプリケーションを実行するには

1. Amazon Managed Service for Apache Flink のコンソールで、My Application を選択し、Run を選択します。

2. 次のページのアプリケーション復元設定ページで、最新のスナップショットで実行 を選択し、 の実行 を選択します。

   アプリケーションのステータスの詳細が、アプリケーションの開始Running時に から Ready に移行Startingし、その後 に移行します。

ダッシュボードを開くには

1. Open Apache Flink ダッシュボード を選択します。ダッシュボードが新しいページで開きます。

2. Runing jobs リストで、表示できる 1 つのジョブを選択します。

   注記

   ランタイムプロパティを設定したり、IAMポリシーを誤って編集したりすると、アプリケーションのステータスが になる可能性がありますがRunning、Flink ダッシュボードにはジョブが継続的に再起動していると表示されます。これは、アプリケーションの設定が間違っているか、外部リソースへのアクセス許可がない場合によく発生する障害シナリオです。

   この場合、Flink ダッシュボードの例外タブをチェックして、問題の原因を確認します。

メトリクスを表示するには

1. 更新ボタンの横にあるドロップダウンリストから 10 秒を選択します。

2. アプリケーションが実行中で正常であれば、アップタイムメトリクスが継続的に増加していることを確認できます。

3. フル再起動メトリクスはゼロである必要があります。増加している場合、設定に問題がある可能性があります。問題を調査するには、Flink ダッシュボードの例外タブを確認します。

4. 正常なアプリケーションでは、失敗したチェックポイントメトリクスの数はゼロである必要があります。

   注記

   このダッシュボードには、5 分の粒度を持つ固定メトリクスのセットが表示されます。 CloudWatch ダッシュボードに任意のメトリクスを含むカスタムアプリケーションダッシュボードを作成できます。

出力を表示するには

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

2. リージョンがこのチュートリアルの実行に使用しているリージョンと同じであることを確認します。デフォルトでは、us-east-1US East (バージニア北部) です。必要に応じてリージョンを変更します。

3. データストリーム を選択します。

4. 監視するストリームを選択します。このチュートリアルでは、ExampleOutputStream を使用します。

5. データビューワータブを選択します。

6. 任意のシャード を選択し、最新の を開始位置 として保持し、レコードの取得 を選択します。このリクエストの「レコードが見つかりません」エラーが表示される場合があります。その場合は、レコードの取得を再試行する を選択します。ストリーム表示に発行される最新のレコード。

7. データ列の値を選択して、レコードの内容を JSON 形式で検査します。

アプリケーションを停止するには

1. アクションドロップダウンリストから、停止 を選択します。

2. アプリケーションのステータスの詳細が から Running に移行しStopping、アプリケーションが完全に停止Readyすると に変わります。

   注記

   Python スクリプトまたは Kinesis Data Generator から入力ストリームへのデータ送信も停止することを忘れないでください。