Kinesis Data Analytics によるニアリアルタイムなデータ分析ことはじめ

「AWS & Game Advent Calendar 2020」 14日目の記事です。

Kinesis Data Analytics利用していますか？
Kinesis Data AnalyticsはAWSが提供するKinesisサービスの1つとなりますが、ストリームデータに対して条件をつけて分析をすることができます。データ分析というと大量のデータに対して集計をかけることや仮説検証のためにアドホックなクエリを実行する話もよく出てくると思いますが、ユーザが行動したタイミングに可能な限り近いタイミングでユーザの行動を把握し対応することができると、ユーザに対して高い効果で影響が与えられるものです。

このようなニアリアルタイムな分析では大量データの分析とは異なる技術スタックが求められ、ストリーミングデータ処理を扱える基盤を整備すると柔軟性のある構成をとることができます。ここではKinesis Data Analyticsの開発/検証が簡単にできることを紹介し、ゲームでの利用ポイントを考えたいと思います。

Kinesis Data Analytics の利用

Kinesis Data Analytics(KDA)ではKinesis Data Streams(KDS)やKinesis Data Firehose(KDF)からデータを受け取り、集計したデータをKDS、KDFやLambdaに送ることができます。
KDSでデータを受け取り、KDAで処理をする流れを見ていきます。

Kinesis Data Streams の作成

まずログデータを受け取るKinesis Data Streamsを作成します。
本番環境で利用する際はシャード数等について検討する必要がありますが、ウィザードに沿って作成をします。

シミュレーションデータ生成

KDSでストリームを作成したら、ここにストリームデータを送付します。
データ送付には Amazon Kinesis Data Generator(KDG) を利用します。

• 用意されているCloudFormationテンプレートを利用してKDG認証用のCognitoの作成(ユーザ名/パスワードを指定)
• テンプレート作成完了後、ページへアクセスし認証を完了
• ストリーミングデータの送り先や生成内容を記載

する事で手軽にシミュレーションデータの生成ができます。
ここで実際に出力されるログ形式をjsonで定義することでKinesis Data Analyticsのアプリケーション作成を進めることができます。

ストリームデータの生成内容はjsonで定義しますが、今回は下記のように定義してデータを生成します。

stream.json

{
    "event_id":{{random.number(10000)}},
    "event_type":{{random.number(100000)}},
    "event_name":"{{random.arrayElement([
        "app_load",
        "login",
        "user_registration",
        "iap_transaction",
        "virtual_goods_transaction"
        ])     
    }}",
    "event_timestamp":{{random.number(10000)}},
    "event_version":{{random.number(50)}},
    "item_id":{{random.number(20)}},
    "item_name":"{{random.weightedArrayElement({
        "weights":[
            0.05,
            0.05,
            0.05,
            0.05,
            0.10,
            0.10,
            0.05,
            0.05,
            0.20,
            0.10,
            0.05,
            0.15
         ],
         "data":[
            "Lavender Hat",
            "Lime Hat",
            "Garnet Hat",
            "Student Hat",
            "Starry Hat",
            "Lava Hat",
            "Glass Hat",
            "Poo Hat",
            "Unicorn Hat",
            "Frost Hat",
            "Rainbow Hat",
            "Love Hat"
         ]
    })}}",
    "item_amount":{{random.number(10000)}},
    "real_value":{{random.number(30)}},
    "virtual_value":{{random.number(10000)}},
    "currency_type":"USD",
    "country_id":"{{random.weightedArrayElement({
         "weights":[
            0.5, 0.2, 0.1, 0.1, 0.1],
            "data":[
            "UNITED STATES",
            "FRANCE",
            "CANADA",
            "MEXICO",
            "SPAIN"]
    })}}",
    "platform":"{{random.weightedArrayElement({
            "weights":[
            0.05, 0.2, 0.1, 0.1, 0.05, 0.5],
            "data":[
            "UnityEditor",
            "Xbox",
            "PlayStation",
            "Android",
            "PC",
            "iPhone"
    ]})}}",
    "transaction_receipt":"e11cebee-806d-4e1d-8bd1-ceb85bda6fdf",
    "application_id":"test-id-here"
}

その後、KDGの画面でストリームデータの送信を開始するとデータ送付がされます。
CloudWatchを確認すると、ストリームデータを受信できていることが確認できますね。

Kinesis Data Analytics(KDA) の作成

KDSにデータが送付されましたので、このストリームデータを受け取り、集計を行うKDAのアプリケーションを作成していきます。
[Create Application]をクリックしてコンソールの通りに進めていき、先ほど作成したKDSのストリームを指定します。
その際に、スキーマ検出機能を利用すると、ストリーム内のデータを見てスキーマを作成できるため便利です。先ほどのシミュレーションデータ生成を実施しながら、進めるとスキーマが検出されました。

そのままコンソールを進めていくとKinesis Data Analyticsの作成が完了します。
作成後、Go to SQL resultsへいくと、ストリームデータに対して、クエリを実行することができます。
このクエリ画面でアドホックなクエリを実行し、仮説と検証をした上で、どのような分析を恒常的に実行するか検討をすすめることができますね。
なお、KDAでDestinationを指定することで、抽出したデータをKinesis Data Streams、Kinesis Data Firehose、Lambdaに出力ができます。抽出されたデータに対して何か評価をしたい場合はLambdaを利用できます。

ここでは次のようなクエリを実行してみます。

stream.json

CREATE OR REPLACE STREAM "data_stream" ("event_id" INTEGER, "event_type" INTEGER, "event_name" varchar(50),
        "event_timestamp" INTEGER, "event_version" INTEGER, "item_id" INTEGER, "item_name" varchar(50), 
        "item_amount" INTEGER, "real_value" INTEGER, "virtual_value" INTEGER, "currency_type" varchar(50), "country_id" varchar(50),
        "platform" varchar(50));
CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "data_stream"
SELECT STREAM "event_id", "event_type", "event_name", "event_timestamp", "event_version", "item_id", "item_name", 
        "item_amount", "real_value", "virtual_value", "currency_type", "country_id",
        "platform"
FROM "SOURCE_SQL_STREAM_001"
WHERE "platform" = 'iPhone';

このようにすることでプラットフォーム列がiPhoneであるストリームを抽出できます。

Kinesis Data Analytics(KDA) の活用

KDAではウインドウクエリを利用することでストリームデータをある程度まとめて処理することができます。

• Stagger Window(ずらしウィンドウ)：パーティションキーに一致する最初のイベントが届いたときにウィンドウが開き、ウィンドウ内で集計
• タンブリングウィンドウ：ストリームを固定時間のウィンドウで区切り、ウィンドウ内で集計
• スライディングウィンドウ：数秒や数分などの固定時間でウィンドウを推移させ、ウィンドウ内で集計

これらを活用することで意味のあるニアリアルタイムな分析をすることができます。

ウィンドウクエリを利用した分析

ずらしウィンドウによるユーザ分析

ずらしウィンドウでは指定したパーティションキーの単位で集計をすることができます。そのため、ユーザIDやイベントIDをパーティションキーと指定することで、特定のユーザがイベントを始めたタイミングでどのような行動をしているか等、ゲーム性に合わせて分析をすることができます。
※何かサンプルができたら更新します。

タンブリングウィンドウによるKPI集計

タンブリングウィンドウではGROUP_BY句を利用して表現をすることができます。
ここではアプリケーション内にストリームが追加されたタイムスタンプであるROWTIMEを用いて5秒単位で集計をしています。ここで売上の値を集計する等して定期的なKPI集計に利用することができます。

CREATE OR  REPLACE STREAM "DESTINATION_SQL_STREAM"(
  "item_name" varchar(50),
  "item_count" INTEGER
  );
CREATE OR  REPLACE PUMP "STREAM_PUMP" AS
  INSERT INTO "DESTINATION_SQL_STREAM"
    SELECT STREAM "item_name", 
                  count("item_name") as "item_count"
    FROM "SOURCE_SQL_STREAM_001"
    GROUP BY "item_name",
      STEP("SOURCE_SQL_STREAM_001".ROWTIME BY INTERVAL '5' SECOND);

このような感じで5秒単位で集計結果が得られます。

KDAで集計したデータはLambdaに連携できるため後続の処理も含めて分析基盤を整備することができるため、任意のダッシュボードへの連携やユーザへの通知までをセットとした環境の用意を進めることができます。ユーザが困っている時や何かを求めている時、出来るだけ早くそのユーザへ応えていけるようなシステムに改善していければいいなと感じます。

おわりに

ここではKinesis Data Analyticsについて触れていきました。シミュレータが用意されていたりクエリ実行をする画面が用意されていたりして手軽に開発を進められるためニアリアルタイムな処理をしたいときに活用をしていきたいところですね。

この記事では下記URLを参考にしました。良い年末を！

https://serverless-game-analytics.workshop.aws/en/
https://aws.amazon.com/jp/blogs/big-data/create-real-time-clickstream-sessions-and-run-analytics-with-amazon-kinesis-data-analytics-aws-glue-and-amazon-athena/