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

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

1. 最初のストリーム (ExampleInputStream) を作成するには、次の Amazon Kinesis create-streamAWS 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. Kinesis コンソール (https://console.aws.amazon.com/kinesis) を開きます。

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

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. にサインインしAWS マネジメントコンソール、https://console.aws.amazon.com/flink で Amazon MSF コンソールを開きます。

2. 正しいリージョンが選択されていることを確認してください: us-east-1 米国東部 (バージニア北部)

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

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

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

   • [Apache Flink の設定、Application Flink バージョン] で [Apache Flink 1.20] を選択します。

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

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

   • [説明] に「My java test app」と入力します。

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

6. [アプリケーション設定用テンプレート] を設定します

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

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

ポリシーを編集する方法

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

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

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

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

   JSON

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

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

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

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

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

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 のコンソールで、[自分のアプリケーション] を選択して [実行] を選択します。

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

   [アプリケーション詳細] の [ステータス] は「Ready」から「Starting」に移行し、アプリケーションが起動されると「Running」に移行します。

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

1. [Open Apache Flink ダッシュボード] を選択します。ダッシュボードは新しいページで開かれます。

2. [実行中のジョブ] リストで、表示されている 1 つのジョブを選択します。

   注記

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

   これが発生した場合、Flink ダッシュボードの [例外] タブを見て、問題の原因を確認してください。

メトリクスを表示する方法

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

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

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

4. 正常なアプリケーションでは、[失敗したチェックポイント数] メトリクスは 0 です。

   注記

   このダッシュボードには、5 分の粒度で一定の一連のメトリクスが表示されます。CloudWatch ダッシュボードで任意のメトリクスを使用してカスタムアプリケーションのダッシュボードを作成できます。

出力の表示方法

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

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

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

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

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

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

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

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

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

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

   注記

   Python スクリプトまたは Kinesis Data Generator から入力ストリームへのデータ送信も必ず停止してください。