AthenaとKinesis Data FirehoseでS3のログをクエリする

Kinesis Data FirehoseでログをS3に流し込み、Athenaでクエリする。カスタムプレフィックスを使ってパーティションを自動登録させる方法。

Takahiro Iwasa

2019-08-31

5 min read

Athenaを使えば、インフラを一切プロビジョニングすることなく、S3に保存されたオブジェクトに対して直接SQLクエリを実行できます。Kinesis Data Firehoseでログを配信し、Athenaでクエリするという構成は、この組み合わせの典型的な使い方の一つです。

System Architecture

Kinesis Data Firehoseの作成

これを可能にしているのが、2019年2月に導入されたカスタムS3プレフィックス機能です。FirehoseはS3オブジェクトキーに対してApache Hive形式のプレフィックスを書き込めるようになり、そのプレフィックスを使ってAthenaのMSCK REPAIR TABLEがパーティションを自動的に作成できます。

Create delivery streamを押し、名前を入力します。

Direct PUT or other sourcesを選択します。

レコード処理の設定はスキップします。

配信先としてAmazon S3を選択します。

プレフィックスとエラープレフィックスを以下の形式で設定します。

Prefix: logs/!{timestamp:'year='yyyy'/month='MM'/day='dd'/hour='HH}/
Error Prefix: error_logs/!{timestamp:'year='yyyy'/month='MM'/day='dd'/hour='HH}/!{firehose:error-output-type}

要件に応じてBuffer sizeとBuffer intervalを調整します。

ストレージコストを削減するためGZIP圧縮を使用します。

配信ストリーム用のIAMロールを作成、または既存のものを選択します。

データのストリーミング

Kinesis Data Firehoseの配信ストリームへプログラムからデータを送るには、AWS SDK for PHP: FirehoseClient#putRecordメソッドが使えます。

1
$client = new FirehoseClient([
2
    'region' => '<AWS_REGION>',
3
    'version' => 'latest',
4
]);
5

6
$data = [
7
    'log_id' => 12345,
8
    'url' => 'https://example.com',
9
];
10

11
$client->putRecord([
12
    'DeliveryStreamName' => '<YOUR_STREAM>',
13
    'Record' => [
14
        'Data' => json_encode($data) . PHP_EOL,
15
    ],
16
]);

Athenaテーブルの作成

AthenaコンソールでCreate table from S3 bucket dataを選択します。

データベース名、テーブル名、そしてFirehoseが使用しているS3パスを入力します。

データフォーマットとしてJSONを指定します。

データ構造に応じてカラムを定義します。

クエリのパフォーマンスを向上させるため、パーティション（例：year/month/day/hour）を設定します。

💡 Tip

パーティショニングによりスキャンされるデータ量が最小限に抑えられ、コストが大幅に削減されます。

パーティションは以下のコマンドで読み込みます。

1
MSCK REPAIR TABLE {TABLE_NAME};

Athenaテーブルへのクエリ

パーティションが設定されていれば、データセット全体をスキャンする代わりに、特定の日付範囲を対象にクエリを実行できます。

1
SELECT
2
  *
3
FROM
4
  table_name
5
WHERE
6
  year = 2019
7
  AND month = 8
8
  AND day = 30
9
LIMIT 10;

ベストプラクティス:

WHERE句には必ずパーティションキーを含める。
不要なスキャンを避けるためLIMIT句を使用する。

まとめ

このパイプラインにより、Kinesis Data Firehose経由でログをS3に配信し、データベースのインフラを一切プロビジョニングすることなくAthenaでクエリ可能な状態にすることができました。ここでポイントになるのがカスタムS3プレフィックス機能です。year=/month=/day=/hour=というプレフィックスの下にオブジェクトを書き込むことで、ログが1時間増えるたびにALTER TABLE ... ADD PARTITIONを実行する必要がなくなり、MSCK REPAIR TABLEがパーティションを自ら発見して登録できるようになります。Firehoseの配信ストリームにおけるGZIP圧縮と、WHERE句によるパーティションプルーニングは相互に作用してストレージとクエリごとのスキャンコストの両方を抑えてくれます。Athenaはスキャンしたバイト数に応じて課金されるため、これが直接コストに響いてきます。データ量が増えるにつれて調整する価値があるのはFirehoseのバッファサイズと間隔で、バッファを大きくすると大量の小さなファイルではなく、より少数で大きめのS3オブジェクトが生成されます。こうしたオブジェクトのほうがAthenaのクエリは速くなる傾向にあります。