本記事では、DynamoDBのTTLで削除されるデータをS3にアーカイブし、そのデータをAthena + Lambdaで分析する仕組みをAWS CDKで実装した過程を紹介します。
マネジメントコンソールからAthenaを使うのは簡単ですが、CDKで再現しようとすると「抽象化された便利さ」を手動で構築する必要があり、想像以上に学びが多い体験となりました。
もともとの目的は、 「アーカイブされたS3データから、先月の利用回数を取得する」 ことでした。
DynamoDBはストレージのコストが地味に高く、古いデータを残し続けるのは非現実的です。
そこで、
という流れに落ち着きました。
Athenaは、S3に保存された構造化データ（JSON, CSV, Parquetなど）に対してSQLでクエリができるサービスです。
サーバレスで、インフラ構築不要、使った分だけ課金されます（1クエリあたり、スキャンしたデータ1TBあたり約5ドル）。
分析頻度が低く、専用のDBを用意するほどでもない場合に最適です。
Athenaを使うには以下の3つが絡んできます：
Athena自体にはテーブルの中身を保存する機能がなく、GlueとS3を橋渡しするイメージです。
保存されているデータ形式（JSON Lines）
{"userId": "abc123", "action": "click", "timestamp": "2025-03-15T12:34:56Z"}
{"userId": "xyz789", "action": "purchase", "timestamp": "2025-03-16T14:12:03Z"}
S3バケット（データ本体）
s3://my-archive-bucket/logs/year=2025/month=03/data.jsonl
Glue Data Catalog（テーブル定義）
CREATE EXTERNAL TABLE logs (
userId string,
action string,
timestamp string
)
PARTITIONED BY (
year string,
month string
)
ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe'
LOCATION 's3://my-archive-bucket/logs/';
AthenaでS3からSQLでデータを抽出するには、マネコンでもCDKでもCreateTableする必要があります。
内部的にはSQLクエリがGlueのテーブル作成APIに変換されて、
Glueというサービスにテーブル定義（データ参照先S3のデータ構造を定義）をしている状態になります。
Athenaのクエリ処理は以下のような順番で進みます：
事前にテーブル構造とS3のロケーションを定義します。
AthenaではS3のディレクトリ階層（year=…, month=…）をWhere条件のように使うため、パーティション情報の登録が必要です。
Glue側にパーティテーションを定義するだけではダメで、一度コマンドを実行させて、実際の参照先S3のディレクトリ構造とテーブル定義を紐づけるコマンドが必要となります。
Athenaでは以下のような非同期な流れでクエリが実行されます：
これで、以下のS3パスがAthenaのパーティションとして認識されるようになります
s3://my-archive-bucket/logs/year=2025/month=03/
これで下記でいうwhere区の部分をGlueが実際のS3の構造と結びつけて認識できるようになる、ということです。
SELECT action, COUNT(*) FROM logs
WHERE year='2025' AND month='03'
GROUP BY action;
Athenaは非同期APIなので、クエリ発行から結果取得まで以下の3ステップが必要です。
Lambdaなどで実行する場合の流れ：
StartQueryExecution — クエリ実行開始GetQueryExecution — ステータス確認（SUCCEEDED or FAILED）GetQueryResults — 結果取得SQLに慣れている人だと直ぐに結果が帰っていこないので、あせるかもしれません。
Athenaの場合、上記のように流れがあり、同期的ではなく、上記3ステップを自身で制御しないといけません。
SELECT action, COUNT(*) FROM logs
WHERE year='2025' AND month='03'
GROUP BY action;
このクエリで、先月のログからアクション別件数を集計できます。
サンプルコードです。
import * as cdk from 'aws-cdk-lib';
import { Construct } from 'constructs';
import { AthenaGlueConstruct } from './athena-glue-construct';
import { ArchiveDataToS3Construct } from './archive-data-to-s3-construct';
import * as dynamodb from 'aws-cdk-lib/aws-dynamodb';
interface MainStackProps extends cdk.StackProps {
projectName: string;
envName: 'dev' | 'stg' | 'prd';
}
export class MainStack extends cdk.Stack {
constructor(scope: Construct, id: string, props: MainStackProps) {
super(scope, id, props);
// Glue定義（Database, Table, S3）
const athenaGlue = new AthenaGlueConstruct(this, 'AthenaGlue', {
projectName: props.projectName,
envName: props.envName,
env: {
account: this.account,
region: this.region,
},
});
// DynamoDBテーブル（ここはサンプル用に適当なTableを作ってる）
const UserChatHistories = new dynamodb.Table(this, 'NintAiUserChatHistories', {
partitionKey: { name: 'userId', type: dynamodb.AttributeType.STRING },
billingMode: dynamodb.BillingMode.PAY_PER_REQUEST,
});
// バッチ処理（Lambda, EventBridge）
new ArchiveDataToS3Construct(this, 'ArchiveDataToS3', {
projectName: props.projectName,
envName: props.envName,
env: {
account: this.account,
region: this.region,
},
userChatHistories: UserChatHistories,
archiveDataToS3Bucket: athenaGlue.archiveDataToS3Bucket,
archiveDatabase: athenaGlue.archiveDatabase,
});
}
}
athena-glue-construct.ts
export class SimpleAthenaGlueSample extends Construct {
constructor(scope: Construct, id: string) {
super(scope, id);
//出力先S3（Athenaのクエリ結果を保存するバケット）
const archiveAnalyticsExportBucket = new s3.Bucket(this, 'ArchiveAnalyticsExportBucket', {
blockPublicAccess: s3.BlockPublicAccess.BLOCK_ALL,
encryption: s3.BucketEncryption.S3_MANAGED,
autoDeleteObjects: true,
removalPolicy: cdk.RemovalPolicy.DESTROY,
objectOwnership: s3.ObjectOwnership.OBJECT_WRITER,
enforceSSL: true,
});
// Glue Database
const database = new glue.CfnDatabase(this, 'SampleDatabase', {
catalogId: cdk.Stack.of(this).account,
databaseInput: { name: 'sample_database' },
});
// Glue Table
new glue.CfnTable(this, 'SampleTable', {
catalogId: cdk.Stack.of(this).account,
databaseName: database.databaseInput.name!,
tableInput: {
name: 'sample_table',
tableType: 'EXTERNAL_TABLE',
parameters: { classification: 'json' },
storageDescriptor: {
columns: [
{ name: 'userId', type: 'string' },
{ name: 'action', type: 'string' },
{ name: 'timestamp', type: 'string' },
],
location: `s3://${bucket.bucketName}/sample-data/`,
inputFormat: 'org.apache.hadoop.mapred.TextInputFormat',
outputFormat: 'org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat',
serdeInfo: {
serializationLibrary: 'org.openx.data.jsonserde.JsonSerDe',
},
},
},
});
}
}
archive-data-to-s3-construct.ts
import * as cdk from 'aws-cdk-lib';
import { Duration } from 'aws-cdk-lib';
import { aws_iam as iam, aws_lambda as lambda, aws_s3 as s3, aws_events as events, aws_events_targets as targets, aws_logs as logs } from 'aws-cdk-lib';
import { NodejsFunction } from 'aws-cdk-lib/aws-lambda-nodejs';
import { Construct } from 'constructs';
interface SimpleArchiveDataToS3Props extends cdk.StackProps {
projectName: string;
envName: 'dev' | 'stg' | 'prd';
env: cdk.Environment;
userChatHistories: cdk.aws_dynamodb.ITable;
archiveDataToS3Bucket: cdk.aws_s3.Bucket;
archiveDatabase: cdk.aws_glue.CfnDatabase;
}
export class SimpleArchiveDataToS3 extends Construct {
constructor(scope: Construct, id: string, props: SimpleArchiveDataToS3Props) {
super(scope, id);
// --- 参照先S3（データが置かれるバケット）
const archiveDataToS3Bucket = new s3.Bucket(this, 'ArchiveDataToS3Bucket', {
blockPublicAccess: s3.BlockPublicAccess.BLOCK_ALL,
encryption: s3.BucketEncryption.S3_MANAGED,
autoDeleteObjects: true,
removalPolicy: cdk.RemovalPolicy.DESTROY,
objectOwnership: s3.ObjectOwnership.OBJECT_WRITER,
enforceSSL: true,
});
// --- Lambda
const archiveDataToS3Function = new NodejsFunction(this, 'ArchiveDataToS3Function', {
runtime: lambda.Runtime.NODEJS_22_X,
entry: './lambda/archive-data-to-s3.ts',
timeout: Duration.minutes(5),
memorySize: 256,
environment: {
ARCHIVE_DATA_TO_S3_BUCKET_NAME: archiveDataToS3Bucket.bucketName,
ATHENA_OUTPUT_BUCKET: archiveAnalyticsExportBucket.bucketName,
ARCHIVE_DATABASE_NAME: props.glueDatabaseName,
},
logGroup: new logs.LogGroup(this, 'ArchiveDataToS3LogGroup', {
retention: logs.RetentionDays.THREE_DAYS,
removalPolicy: cdk.RemovalPolicy.DESTROY,
}),
});
// --- IAM設定
// DynamoDBアクセス許可
archiveDataToS3Function.addToRolePolicy(new iam.PolicyStatement({
actions: ['dynamodb:Scan', 'dynamodb:Query'],
resources: props.userChatHistories,
}));
// LambdaにS3参照・書き込みを許可（参照先）
archiveDataToS3Function.addToRolePolicy(new iam.PolicyStatement({
actions: ['s3:GetObject', 's3:PutObject', 's3:ListBucket', 's3:GetBucketLocation'],
resources: [
archiveDataToS3Bucket.bucketArn,
archiveDataToS3Bucket.arnForObjects('*'),
],
}));
// LambdaにS3参照・書き込みを許可（出力先）
archiveDataToS3Function.addToRolePolicy(new iam.PolicyStatement({
actions: ['s3:PutObject', 's3:GetBucketLocation'],
resources: [
archiveAnalyticsExportBucket.bucketArn,
archiveAnalyticsExportBucket.arnForObjects('*'),
],
}));
// LambdaにAthenaアクセス許可
archiveDataToS3Function.addToRolePolicy(new iam.PolicyStatement({
actions: [
'athena:StartQueryExecution',
'athena:GetQueryExecution',
'athena:GetQueryResults',
],
resources: ['*'],
}));
// LambdaにGlueアクセス許可
archiveDataToS3Function.addToRolePolicy(new iam.PolicyStatement({
actions: [
'glue:GetDatabase',
'glue:GetDatabases',
'glue:GetTable',
'glue:GetTables',
'glue:UpdateTable',
'glue:CreateTable',
'glue:BatchCreatePartition',
'glue:BatchDeletePartition',
'glue:GetPartition',
'glue:GetPartitions',
'glue:UpdatePartition',
'glue:BatchGetPartition',
],
resources: [
`arn:aws:glue:${cdk.Stack.of(this).region}:${cdk.Stack.of(this).account}:catalog`,
`arn:aws:glue:${cdk.Stack.of(this).region}:${cdk.Stack.of(this).account}:database/${props.glueDatabaseName}`,
`arn:aws:glue:${cdk.Stack.of(this).region}:${cdk.Stack.of(this).account}:table/${props.glueDatabaseName}/*`,
],
}));
// AthenaにS3アクセス権限を付与（出力先）
prop.archiveAnalyticsExportBucket.addToResourcePolicy(new iam.PolicyStatement({
principals: [new iam.ServicePrincipal('athena.amazonaws.com')],
actions: ['s3:GetObject', 's3:PutObject', 's3:GetBucketLocation'],
resources: [
archiveAnalyticsExportBucket.bucketArn,
archiveAnalyticsExportBucket.arnForObjects('*'),
],
}));
// AthenaにS3アクセス権限を付与（参照先）
archiveDataToS3Bucket.addToResourcePolicy(new iam.PolicyStatement({
principals: [new iam.ServicePrincipal('athena.amazonaws.com')],
actions: ['s3:GetObject', 's3:GetBucketLocation', 's3:ListBucket'],
resources: [
archiveDataToS3Bucket.bucketArn,
archiveDataToS3Bucket.arnForObjects('*'),
],
}));
// --- EventBridgeスケジュール
new events.Rule(this, 'ArchiveDataToS3ScheduleRule', {
schedule: events.Schedule.cron({ minute: '0', hour: '18', day: '1', month: '*', year: '*' }),
targets: [new targets.LambdaFunction(archiveDataToS3Function)],
});
}
}
Athenaは軽量なデータや少ないリクエスト回数であればかなりコストやすい
Glueは100万オブジェクトの定義、100万リクエストまで無料なので、コストが殆どかからないことです。
Athena・Glue・S3すべてに対してLambdaからアクセスできるようにするIAM設計が必要です。
CDKでも記述量が多く、最初は少し混乱しました。
例えば、作成したラムダとS３にはこのような権限を与えています。
これだけ必要で結構煩雑になってしまいます。
一番厄介なのがGlueで、LambdaからAthenaクエリを実行するとき、Athenaがクエリを内部的にGlueのコマンドに変えているので、どのようなコマンドを投げているのかわかりません。
ただ、LambdaからAthenaを利用している以上、実行権限が必要なので、IAMが煩雑になってしまいます。
必要なポリシーだけつけたいのですが、また厄介なのが権限エラーもクエリエラーも、Athenaがクエリエラーと判定してしまうところにあります。
Athena returns "FAILED: Execution Error, return code 1 from org.apache.hadoop.hive.ql.exec.DDLTask. null"
これが気をつけるポイントです。
Athenaは同期的なSQL実行ではないため、Lambda内でポーリング処理や完了確認ロジックが必要になります。
AthenaでSQLでデータを取得するにあたって、順番が非常に大事になるからです。
と順番がとても大事。
マネコンだと裏でよしなにやってくれますが、CDKでやるとなると、順序や依存関係の意識が必須になります。
CDKでAthenaを使うのは、「抽象化されたマネコンの便利さを一つ一つ具現化していく作業」に近いです。
その分、サービス理解は格段に深まりました。
仕組みを自動化・明文化できたことで、チーム内でも共有しやすくなったのは大きなメリットです。
AthenaをLambdaで使うにあたって手順が思ったより多かったのでStep Functionをリファクタリング時には検討しようと思います。
AWS Step Functionsは、複数の処理（Lambdaとかバッチとか）を順番に実行したり、条件分岐したり、エラー処理を組み込んだりできるサービス。
簡単に言うと、「AWS上でフロー（ワークフロー）を作るツール」です。
しかも、全部マネージドサービスだから
・並列実行
・タイムアウト
・リトライ
成功・失敗の分岐
とかが簡単に組めるので楽になると思います。
2025年4月現在、@aws-cdk/aws-glue-alpha というL2コンストラクトが出ています。
これを使うと、Glue周りのコードがより簡潔に書けるようになるかもしれません。
今回、Athena + Lambda + CDK という組み合わせで、
低コストでデータ分析を自動化する構成を作ってみました。
一見シンプルに見える構成ですが、実際に手を動かしてみると
実際、インフラでログ管理する人や、データエンジニア界隈ではこの構成はかなりよく使われているようで、記事もよく見かけます。
シンプルながら実戦的な設計力を試される構成だと改めて感じました。
それでも、こういう実用的な構成を自分の手で作れる経験は、
AWS力アップにも直結するので、「チャレンジしたい人」にはぜひおすすめしたいです！☺️👍
Be the first to comment
