Kinesis Data Streams から Kinesis Data Firehose の Dynamic Partitioning を使って S3 に格納してみた

はじめに

AWS でストリームデータを活用する一つの方法として、Kinesis Data Streams → Kinesis Data Firehose → S3 → Athena → QuickSight という組み合わせがあります。受け取ったストリームデータを S3 に格納し、Athena と QuickSight で可視化をする構成です。Athena では S3 上に格納された、JSON, CSV, Parquet といったデータをスキャンして集計のクエリーを発行できます。Athena の料金の考え方のひとつとして、スキャン料金があります。S3 上に格納されたデータをスキャンした容量によって料金が発生します。たとえば、毎日の単位でディレクトリを分けることで、特定の期間に限定したスキャンが出来るパーティションと呼ばれる機能があります。

利用のイメージはこんなかんじです。

• https://www.slideshare.net/AmazonWebServicesJapan/20200617-aws-black-belt-online-seminar-amazon-athena

大事なポイントは、S3 にデータを格納するときに、パーティションを意識してディレクトリを区切ったうえで、データを格納しないといけない点があります。これを自分たちでプログラムを作成していくのは面倒なので、Kinesis Data Firehose に自動的にディレクトリを切ってくれる機能があります。

ただ、何も意識しないで利用すると、 Kinesis Data Firehose が受信した UTC 時刻を使って ディレクトリが自動的に作成されます。UTC 時刻なので、日本時間と時間差が生じることになります。具体例を見てみましょう。

JSON データを Kinesis Data Streams, Kinesis Data Firehose を経由して S3 に格納している図を表現しています。日付に関して何も意識せず利用すると、UTC 時刻を使います。そのため、日本時刻の 2022-09-02 08:00 に受信したデータは、UTC 的に 2022-09-01 のディレクトリに格納されます。この問題点は、Athena をつかって SQL クエリーを実行しても、正確な時刻でデータを取得できない問題点があります。

これを解決するために、Kinesis Data Firehose の Dynamic Partitioning という機能を利用できます。

Dynamic Partitioning を使うことで、受け取った JSON データの中身に記録されている日付を確認し、その日付に合わせたディレクトリに格納することができます。これによって、Athena の SQL クエリーの時刻指定も、より厳密な期間を指定しやすくなります。

なお、Dynamic Partitioning の機能は、通常の Kinesis Data Firehose の料金に加えて、追加料金が発生します。具体的なものは料金ページをご覧ください。

• https://aws.amazon.com/jp/kinesis/data-firehose/pricing/

それでは、Dynamic Partitioning の設定方法を確かめていきましょう！

Kinesis Data Firehose を作成

Kinesis の画面に移動して、delivery stream を作成していきます。メニューに「Firehose」というキーワードは出てこないです。Kinesis Data Firehose のサービスで提供されている delivery stream を作っていく考え方ですね。

データの入力元 Source の指定、出力先 Destination の指定を行います。今回は、入力元を Kinesis Data Streams にして、出力先を S3 にしていきます。

データの変更や加工はしません

以下のパラメータを入れます。

• 出力先に S3 Bucket を指定
• Dynamic partitioning を有効
• New line delimiter を有効化
  • Kinesis Data Firehose では、S3 にデータを保存するときに、複数のデータをまとめて出力先に保存します。まとめる単位は、「一定時間ごと」もしくは「一定容量ごと」にまとめられます。New Line delimiter を有効にすると、複数のレコードの間に自動的に改行を入れてくれるため、基本的には有効化が良いと思います。(この機能を無効化すると、1行に複数のデータが入れられるので逆にやりにくい)

New line delimiter について、具体的な例を上げます。無効化のときは、このように2つのレコードがまとまって1行となっています。

{"type":"clickEvent","date":"2022-09-18 13:47:34","url":"http://test.com","contentsName":"すごいコンテンツ1"}{"type":"clickEvent","date":"2022-09-18 13:47:34","url":"http://test.com2","contentsName":"すごいコンテンツ2"}

New line delimiter を 有効化したときには、複数のレコード間で自動的に改行されます。Athena などから利用する際には別々のレコードは改行されていることが期待されているので、New line delimiter は有効化の方が便利だと思います。

{"type":"clickEvent","date":"2022-09-18 13:47:34","url":"http://test.com","contentsName":"すごいコンテンツ1"}
{"type":"clickEvent","date":"2022-09-18 13:47:34","url":"http://test.com2","contentsName":"すごいコンテンツ2"}

次に、Dynamic Partitioning の動作を指定していくのですが、データのフォーマットを意識する必要があります。この記事の環境では、この Kinesis Data Firehose に流すデータのフォーマットは、次の形式になっています。

{"type":"clickEvent","date":"2022-09-17 20:43:33","url":"http://test.com","contentsName":"すごいコンテンツ"}

このフォーマットを基に、マネージメントコンソールにパーサーのための文字列を入力する必要があります。例えばこんなかんじです。

Dynamic Partitioning の裏側では、Linux などで利用できる jq コマンドが使われています。jq コマンドは、JSON メッセージから抽出などが行える便利なパーサーコマンドです。上記の画像の JQ expression は、jq コマンドとして扱うための正規表現です。

この正規表現を、マネージメントコンソール上で動作確認するのは大変なので、ローカルの環境でまずは動作確認を行います。なお、jq コマンドで利用できる function などはこちらに記載されています

• https://stedolan.github.io/jq/manual/#Dates

上記の Document を見ながら、このように jq コマンドとしてパーサーできることがわかります。

echo -n '{"type":"clickEvent","date":"2022-09-17 20:43:33","url":"http://test.com","contentsName":"すごいコンテンツ"}' | jq '.date | strptime("%Y-%m-%d %H:%M:%S")'

ポイントを整理すると、Kinesis Data Firehose に流れくる時刻のデータは "date":"2022-09-17 20:43:33" です。これを jq 上で Date 型に変換するために .date | strptime("%Y-%m-%d %H:%M:%S") を使っています。

実行例

> echo -n '{"type":"clickEvent","date":"2022-09-17 20:43:33","url":"http://test.com","contentsName":"すごいコンテ
ンツ"}' | jq '.date | strptime("%Y-%m-%d %H:%M:%S")'
[
  2022,
  8,
  17,
  20,
  43,
  33,
  6,
  259
]

この変換を踏まえて、JQ Expression を入力していきます。

• year という変数に、年数を入れる
• month という変数に、月数を入れる
• day という変数に、日数を入れる

.date | strptime("%Y-%m-%d %H:%M:%S") | strftime("%Y")
.date | strptime("%Y-%m-%d %H:%M:%S") | strftime("%m")
.date | strptime("%Y-%m-%d %H:%M:%S") | strftime("%d")

そして、実際に S3 Bucket に出力するディレクトリ構成を指定します。上記で指定した year・month・day の変数を使って、S3 Prefix を指定します。partition_date=年-月-日/ の形式で、S3 に保存する形式です。これは、Athena の Partition Projection の形式を意識しています。

partition_date=!{partitionKeyFromQuery:year}-!{partitionKeyFromQuery:month}-!{partitionKeyFromQuery:day}/

バッファーなどはデフォルトのままです。

Create を押します。

作成中となります。

一定時間後、作成が完了されます。

動作確認

データを流して、実際に S3 Bucket に正しいディレクトリ構成で出力されるか確認をしていきます。データの流し方の詳細は省略します。(JavaScript のプログラムから流します)

node clickcount.js

一定時間後、S3 にディレクトリが切られデータが格納されています！素晴らしい！

このディレクトリの中身は、複数の JSON ファイルが格納されています。

検証を通じてわかったこと

• Kinesis Data Firehose には、**「Firehose がデータを受け取った UTC 時刻」**を基にプレフィックスを付与してくれる機能がある。

  • ただ、日本時刻で考えたときに、若干わかりにくい
  • UTC 上だと 9/15 だけど、実際の日本時刻だと 9/16 のデータなことが起こりえる
  • https://docs.aws.amazon.com/ja_jp/firehose/latest/dev/s3-prefixes.html

• 上記の懸念を解決するために Dynamic Partitioning を活用可能

• Kinesis Data Firehose で Dynamic Partitioning を有効化し、その中にある New line delimiter の機能を活用することで、複数のレコードを改行することが出来る。

  • New line delimiter について、具体的な例を上げます。無効化のときは、このように2つのレコードがまとまって1行となっています。

    {"type":"clickEvent","date":"2022-09-18 13:47:34","url":"http://test.com2","contentsName":"すごいコンテンツ2"}{"type":"clickEvent","date":"2022-09-18 13:47:34","url":"http://test.com","contentsName":"すごいコンテンツ"}
    

    有効化のときは改行されます。Athena などから利用する際には別々のレコードは改行されていることが期待されているので、New line delimiter は有効化の方が便利だと思います。

    {"type":"clickEvent","date":"2022-09-18 13:47:34","url":"http://test.com2","contentsName":"すごいコンテンツ2"}
    {"type":"clickEvent","date":"2022-09-18 13:47:34","url":"http://test.com","contentsName":"すごいコンテンツ"}
    

• Dynamic Partitioning の裏側で、jq コマンドが使われている。バージョンは 1.6

  • マネージメントコンソールにバージョンも書かれている
    

• Dynamic Partitioning は、後から有効化することができない。Delivery Stream を作成したときに指定する必要がある。

• Dynamic Partitioning は追加の料金が発生する

  • https://aws.amazon.com/jp/kinesis/data-firehose/pricing/

参考URL