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

![image-20220920224448478.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/365f6c9e-ee7f-fd02-0749-c933c8e1dc2d.png)



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

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

![image-20220920230258300.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/2d304262-429a-bf14-a85c-51568c951a07.png)

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 という機能を利用できます。

![image-20220920230628246.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/fb136683-d25b-37ad-c696-aff53e1bf066.png)



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 を作っていく考え方ですね。

![image-20220918113329754.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/64436b78-fc84-081e-160e-3837a7d044f6.png)



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

![image-20220918131201074.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/d57819ba-e18c-5546-5fbd-9b5dfc1439df.png)



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

![image-20220918131454829.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/76779ded-b537-9ac6-d7a0-070a6627bce6.png)



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

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

![image-20220918140449168.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/6229f3a5-f1de-30c9-bfa1-fc4905534ea3.png)



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":"すごいコンテンツ"}
```



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


![image-20220918133855448.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/225c1cb7-b270-eab4-818d-a7a30e4f2393.png)



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")
```

![image-20220918133855449.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/27ab6fa1-c649-aa93-a59d-94f1dcd14571.png)



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

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

![image-20220918174834125.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/e4876aa3-bd33-319e-2078-b67520f38ae9.png)



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

![image-20220918134344113.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/61df2748-827f-a6da-f459-8a7409703680.png)



Create を押します。

![image-20220918134354701.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/bce21fc3-0b1c-13db-59a0-95dc1d0e95e7.png)



作成中となります。

![image-20220918134449490.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/f287c91c-6666-236c-27b0-2b609dfa5c7e.png)



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

![image-20220918134519920.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/c1e13c70-a15c-4427-d01b-5dd5edf1cba5.png)



# 動作確認

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

```
node clickcount.js
```



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

![image-20220920233514441.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/7abca42b-c5b9-bcdb-9609-1407eb026e72.png)



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

![image-20220920233536633.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/1c148eb7-d007-41fa-a6e5-3fd706bd35de.png)



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

- Kinesis Data Firehose には、**<u>「Firehose がデータを受け取った UTC 時刻」</u>**を基にプレフィックスを付与してくれる機能がある。
  - ただ、日本時刻で考えたときに、若干わかりにくい
  - 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

  - マネージメントコンソールにバージョンも書かれている
![image-20220920232615774.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/835f5d9d-db07-3daa-c5c9-650e73dc4c8b.png)

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

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

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



# 参考URL

https://qiita.com/kkkdev/items/e4c931bc54dcf3442a0b