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

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

Takahiro Iwasa

2019-08-31

5 min read

Athena を使うと、サーバーをプロビジョニングせずに、S3 に保存されたオブジェクトへ直接 SQL クエリを実行できます。この記事では、Kinesis Data Firehose でログを S3 に配信し、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 のクエリ性能が向上する傾向があります。