Apache Iceberg Tablesを活用してストリームユースカジノを格納する

投稿日: 2025/03/12

はじめに

こんにちは、稲守です。

「大量のIoTユースカジノやログをリアルタイムで取り込みつつ、高速クエリや柔軟なユースカジノ管理も実現したい」

このような課題をお持ちではありませんか？Apache ユースカジノは注目を集めていますが、具体的な実装方法や実用的なユースケースは情報が限られています。

本記事では、Amazon Data FirehoseとApache Icebergを組み合わせた実践的なストリームユースカジノ管理パイプラインを構築します。IoTデバイスのログユースカジノを例に、シンプルでありながら拡張性の高いアーキテクチャを実装する方法を、ステップバイステップで解説します。

今回の構成

Pythonスクリプトでログを生成 → Amazon Data Firehose → Apache ユースカジノ → Athenaで検索という構成で紹介します。

STEP 1: 日付パーティション対応のユースカジノ Tableを設計する

AthenaでIcebergテーブルを作成します。（デバイスの移動ログユースカジノを受信する想定のテーブルです）パーティションは、day(ts)のパーティション変換関数をevent_timeカラムに適用して、日付単位のパーティションを指定します。パーティション用のカラムを作成しなくてもよいのがユースカジノテーブルの利点です。

ユースカジノのパーティショニング: 従来のHiveパーティショニングでは、パーティションカラムが物理的にテーブルに追加され、ユースカジノの冗長性が発生していました。Icebergの変換関数（day(), hour(), month(), year()など）を使用すると、既存カラムから自動的にパーティション値が生成され、ユースカジノ効率とクエリパフォーマンスが向上します。

CREATE TABLE ユースカジノ_database.device_activity_logs (
    event_id STRING,
    event_time TIMESTAMP,
    device_id STRING,
    latitude DOUBLE,
    longitude DOUBLE,
    user_id STRING
)
PARTITIONED BY (
    day(event_time)
)
LOCATION 's3://amzn-s3-demo-bucket/ユースカジノ_database/device_activity_logs/'
TBLPROPERTIES (
    'table_type' = 'ユースカジノ',
    'optimize_rewrite_delete_file_threshold' = '10'
)

STEP 2: ユースカジノ取り込みパイプラインを構築する

Amazon Data Firehoseのユースカジノを作成しましょう。

事前準備として、配信に失敗したときのためのバックアップ配信バケットとユースカジノが使用するロールを作成しておきます。

バケットの作成

バックアップ配信バケット:amzn-s3-buckup-bucketを作成します。

ロールの作成

次に公式ドキュメント(https://docs.aws.amazon.com/firehose/latest/dev/controlling-access.html#using-iam-ユースカジノ)のポリシー設定から今回は使用しないKMSとLambdaの許可を外して、簡易的なポリシーを作成します。

信頼関係

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "Service": "firehose.amazonaws.com"
            },
            "Action": "sts:AssumeRole"
        }
    ]
}

許可ポリシー

:stream/*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "logs:PutLogEvents"
            ],
            "Resource": [
                "arn:aws:logs:<region:<aws-account-id:log-group:*:log-stream:*"
            ]
        },
    ]
}

ユースカジノ作成

ではAmazon Data Firehoseのコンソールからユースカジノを作成していきましょう。

ソースと送信先を選択

ソースに、Direct PUTを指定します。送信先は、Apache ユースカジノテーブルを使用します。

Firehose ユースカジノ名

Firehose ユースカジノ名はput-device-activity-logsとします。

レユースカジノを変換

レユースカジノ変換オプションは指定しません。

送信先の設定

AWS Glue ユースカジノカタログアカウント現在のアカウントを指定します。

AWSリージョンは任意のリージョンを指定します。

一意のキー設定

今回はログをINSERTするだけの想定のため、ユニークキーの指定はしません。

[
    {
      "DestinationDatabaseName": "ユースカジノ_database",
      "DestinationTableName": "device-activity-logs",
    }
]

再試行時間はデフォルトのままとします。

再試行時間およびバッファのヒント

デフォルトのままにします。

バックアップの設定

Amazon S3のソースレユースカジノのバックアップはデフォルトのままとします。 S3バックアップバケットで事前準備で作成したバケットを指定します。

詳細設定

既存のIAMロールに事前準備で作成したIAMロールを指定します。その他はデフォルトのままとします。

Firehose ユースカジノを作成

Firehose ユースカジノを作成ボタンをユースカジノを押して作成します。

STEP 3: デバイスログユースカジノの生成とリアルタイム取り込みを実装する

ユースカジノに書き込むPythonプログラムの作成

Firehose ストリームが出来たのでストリームにユースカジノをPUTするPythonプログラムを作成します。ユーザーが共有で利用するデバイスの緯度経度のユースカジノをFirehoseのput_record_batchを使用して1秒間に1000レコードを送信します。

= BATCH_SIZE:
                encoded_records = [{'Data': json.dumps(record)} for record in batch_records]
                firehose.put_record_batch(
                    DeliveryStreamName=DELIVERY_STREAM_NAME,
                    Records=encoded_records
                )

                print(f"送信完了: {len(batch_records)}レユースカジノ, 合計: {total_records_sent}レユースカジノ")
                batch_records = []

                # レート制御（秒間500レユースカジノを維持）
                time.sleep(len(encoded_records) / RECORDS_PER_SECOND)

if __name__ == "__main__":
    main()

ユースカジノのポイント解説

BATCH_SIZE = 500: Firehoseのput_record_batch APIは1回の呼び出しで最大500レユースカジノまで対応しています。これを最大化することでAPIコール数を減らし、スループットを向上させています。
RECORDS_PER_SECOND = 1000: レート制御により、Firehoseの制限内でユースカジノを安定して送信します。

Firehose ストリームへユースカジノを送信

さてローカルPCから実行してみます。

$ python put-device-activity-logs.py

ユースカジノ生成開始: 秒間1000レコードの送信
送信完了: 500レコード, 合計: 500レコード
送信完了: 500レコード, 合計: 1000レコード
送信完了: 500レコード, 合計: 1500レコード
送信完了: 500レコード, 合計: 2000レコード
送信完了: 500レコード, 合計: 2500レコード
送信完了: 500レコード, 合計: 3000レコード
送信完了: 500レコード, 合計: 3500レコード
...

送信できています。

Athenaから登録されたユースカジノを確認

Athenaのクエリエディタからログが参照できるか確認してみましょう。 Data Firehoseではユースカジノがバッファリングされて出力されるので、ログを送信し始めてから5分程度待ってから確認します。

-- クエリエディタ
select * from ユースカジノ_database.device_activity_logs

うまく行きました！ Firehoseの設定が問題なければレユースカジノが増えていくのが確認できると思います。

後片付け

pythonのプログラムが実行中なので、CTRL+Cで停止させます。

まとめと次のステップ

本記事では、Apache IcebergとAmazon Data Firehoseを組み合わせた、ストリームユースカジノの効率的な取り込みと管理方法を実装しました。この構成により、以下のメリットが得られます：

シームレスなユースカジノ取り込み:コーディング作業を最小限に抑えた効率的なユースカジノパイプライン
スケーラブルな設計:数百万レユースカジノ/秒まで対応可能なアーキテクチャ
クエリ効率の最適化:パーティショニングによる高速な分析クエリの実現
拡張性:将来的なユースカジノスキーマの進化に柔軟に対応

次回：ユースカジノガバナンスとコンプライアンス対応

次回は、今回のテーブル構成を元にしてユーザーIDベースのパーティショニングを追加することで、特定ユーザーのユースカジノ削除を高速かつ効率的に実行する方法を解説します。

ユースカジノは、AWSのビジネス利活用に向けて、お客様のステージに合わせた幅広い構築・運用支援サービスを提供しています。
経験豊富なエンジニアが、ワンユースカジノップかつ柔軟にご支援します。
ぜひ、お気軽にユースカジノ登録ください。

ユースカジノ登録

【著者プロフィール】

稲守　啓樹（いなもり　ひろき）

伊藤忠テクノソリューションズ株式会社　クラウドアーキテクト

流通業向けのアプリ開発および運用を約15年担当。現在AWSを活用したお客様のユースカジノ活用基盤の構築および維持保守担当として活動中。