例: Kinesis Data Analytics アプリケーションにリファレンスデータを追加する

テキストエディタを開き、以下のデータを追加して、ファイルを TickerReference.csv として保存します。

Ticker, Company
AMZN,Amazon
ASD, SomeCompanyA
MMB, SomeCompanyB
WAS,  SomeCompanyC

TickerReference.csv ファイルを S3 バケットにアップロードします。手順については、「Amazon Simple Storage Service ユーザーガイド」の「Amazon S3 へのオブジェクトのアップロード」を参照してください。

AWS Identity and Access Management (IAM) で、KinesisAnalytics-ReadS3Object という名前の IAM ロールを作成します。ロールを作成するには、IAM ユーザーガイドにある「Amazon Service (AWS Management Console) 用のロールを作成する」の手順に従ってください。

IAM コンソールで、以下を指定します。

IAM; ロールポリシーを更新します。

1. IAM; コンソールで、作成したロールを選択します。

2. [Trust Relationships (信頼関係)] タブで、信頼ポリシーを更新してロールを引き受ける権限を Kinesis Data Analytics に付与します。以下に信頼ポリシーを示します。

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Effect": "Allow",
         "Principal": {
           "Service": "kinesisanalytics.amazonaws.com"
         },
         "Action": "sts:AssumeRole"
       }
     ]
   }

3. [Permissions] タブで、[AmazonS3ReadOnlyAccess] という Amazon 管理ポリシーをアタッチします。これにより、Amazon S3 オブジェクトを読み取るアクセス権限をロールに付与します。このポリシーを以下に示します。

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Effect": "Allow",
         "Action": [
           "s3:Get*",
           "s3:List*"
         ],
         "Resource": "*"
       }
     ]
   }

アプリケーションのメインページで [Connect reference data (リファレンスデータの接続)] を選択します。

[リファレンスデータソースの接続] ページで、リファレンスデータオブジェクトを含む Amazon S3 バケットを選択し、オブジェクトのキー名を入力します。

[アプリケーション内リファレンステーブル名] に CompanyName と入力します。

[選択したリソースへのアクセス] セクションで、[Kinesis Analytics が引き受ける IAM ロール] を選択後、前のセクションで作成した IAM ロール [KinesisAnalytics-ReadS3Object] を選択します。

[スキーマの検出] を選択します。コンソールによって、リファレンスデータの 2 つの列が検出されます。

[保存して閉じる] を選択します。

アプリケーションコードを以下に置き換えます。クエリはアプリケーション内入力ストリームをアプリケーション内リファレンステーブルと結合します。アプリケーションコードは、結果を別のアプリケーション内ストリーム、DESTINATION_SQL_STREAM に書き込みます。

CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (ticker_symbol VARCHAR(4), "Company" varchar(20), sector VARCHAR(12), change DOUBLE, price DOUBLE);

CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM"
   SELECT STREAM ticker_symbol, "c"."Company", sector, change, price
   FROM "SOURCE_SQL_STREAM_001" LEFT JOIN "CompanyName" as "c"
   ON "SOURCE_SQL_STREAM_001".ticker_symbol = "c"."Ticker";

アプリケーション出力が [SQLResults] タブに表示されていることを確認します。一部の行に会社名が表示されることを確認します (サンプルのリファレンスデータには、すべての会社名はありません)。