S3にCSVを置くだけでDynamoDBにマスターデータを投入する仕組みをAWS CDKで構築してみた

facilityCode,facilityName,prefecture,isActive
TES01,テスト練習場01,東京都,true
TES02,テスト練習場02,神奈川県,true
TES03,テスト練習場03,大阪府,false

csv-import-cdk/
├── lib/
│   └── csv-import-stack.ts   # CDKスタック
├── lambda/
│   └── import-handler.ts     # 取り込み用Lambda
├── bin/
│   └── csv-import.ts         # エントリーポイント
└── package.json

npm install papaparse
npm install -D @types/papaparse @types/aws-lambda

import * as cdk from "aws-cdk-lib";
import {
  aws_dynamodb as dynamodb,
  aws_s3 as s3,
  aws_s3_notifications as s3n,
} from "aws-cdk-lib";
import { Runtime } from "aws-cdk-lib/aws-lambda";
import { NodejsFunction } from "aws-cdk-lib/aws-lambda-nodejs";
import { Construct } from "constructs";

export class CsvImportStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    // 1. マスターデータを格納する DynamoDB テーブル
    const facilityTable = new dynamodb.Table(this, "FacilityTable", {
      tableName: "facility",
      partitionKey: {
        name: "facilityCode",
        type: dynamodb.AttributeType.STRING,
      },
      billingMode: dynamodb.BillingMode.PAY_PER_REQUEST,
      removalPolicy: cdk.RemovalPolicy.DESTROY, // 検証用
    });

    // 2. CSV をアップロードする S3 バケット
    const bucket = new s3.Bucket(this, "CsvImportBucket", {
      blockPublicAccess: s3.BlockPublicAccess.BLOCK_ALL,
      encryption: s3.BucketEncryption.S3_MANAGED,
      removalPolicy: cdk.RemovalPolicy.DESTROY, // 検証用
      autoDeleteObjects: true, // 検証用
    });

    // 3. 取り込み用 Lambda
    const importFn = new NodejsFunction(this, "CsvImportFunction", {
      runtime: Runtime.NODEJS_22_X,
      entry: "lambda/import-handler.ts",
      handler: "handler",
      timeout: cdk.Duration.seconds(60),
      environment: {
        FACILITY_TABLE_NAME: facilityTable.tableName,
      },
    });

    // 4. 権限付与（Lambda → S3 / DynamoDB）
    bucket.grantReadWrite(importFn);
    facilityTable.grantWriteData(importFn);

    // 5. S3 イベント通知
    //    uploads/ 配下に .csv が作成されたら Lambda を起動する
    bucket.addEventNotification(
      s3.EventType.OBJECT_CREATED,
      new s3n.LambdaDestination(importFn),
      { prefix: "uploads/", suffix: ".csv" },
    );
  }
}

{ prefix: "uploads/", suffix: ".csv" }

import {
  CopyObjectCommand,
  DeleteObjectCommand,
  GetObjectCommand,
  S3Client,
} from "@aws-sdk/client-s3";
import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
import {
  BatchWriteCommand,
  DynamoDBDocumentClient,
} from "@aws-sdk/lib-dynamodb";
import type { S3Handler } from "aws-lambda";
import Papa from "papaparse";

const s3 = new S3Client({});
const ddb = DynamoDBDocumentClient.from(new DynamoDBClient({}));

const TABLE_NAME = process.env.FACILITY_TABLE_NAME!;

// CSV の1行（CSV から読むとすべて文字列で入ってくる）
type FacilityCsvRow = {
  facilityCode: string;
  facilityName: string;
  prefecture: string;
  isActive: string;
};

export const handler: S3Handler = async (event) => {
  const record = event.Records[0];
  const bucket = record.s3.bucket.name;
  const objectKey = decodeURIComponent(
    record.s3.object.key.replace(/\+/g, " "),
  );

  const timestamp = new Date().toISOString().replace(/[-:T]/g, "").slice(0, 14);
  const fileName = objectKey.split("/").pop() ?? "facilities.csv";

  try {
    // 1. S3 から CSV を取得
    const obj = await s3.send(
      new GetObjectCommand({ Bucket: bucket, Key: objectKey }),
    );
    const csv = await obj.Body!.transformToString("utf-8");

    // 2. CSV をパース
    const { data } = Papa.parse<FacilityCsvRow>(csv, {
      header: true,
      skipEmptyLines: true,
    });

    // 3. DynamoDB の Item に変換
    //    CSV は文字列なので、isActive はここで boolean に直す
    const now = new Date().toISOString();
    const items = data.map((row) => ({
      facilityCode: row.facilityCode,
      facilityName: row.facilityName,
      prefecture: row.prefecture,
      isActive: row.isActive.toLowerCase() === "true",
      updatedAt: now,
    }));

    // 4. DynamoDB へ書き込み（BatchWrite は 25 件ずつ）
    for (let i = 0; i < items.length; i += 25) {
      const chunk = items.slice(i, i + 25);
      await ddb.send(
        new BatchWriteCommand({
          RequestItems: {
            [TABLE_NAME]: chunk.map((item) => ({
              PutRequest: { Item: item },
            })),
          },
        }),
      );
    }

    // 5. 成功したら processed/ に移動
    await moveFile(bucket, objectKey, `processed/${timestamp}_${fileName}`);
    console.log(`取り込み完了: ${items.length} 件`);
  } catch (error) {
    // 失敗したら error/ に移動して、エラーは投げ直す
    await moveFile(bucket, objectKey, `error/${timestamp}_${fileName}`);
    throw error;
  }
};

// S3 には rename がないので、コピー＋削除でファイルを移動する
const moveFile = async (bucket: string, from: string, to: string) => {
  // CopySource はURLエンコードが必要（スラッシュは保持する）
  const encodedSource = from.split("/").map(encodeURIComponent).join("/");
  await s3.send(
    new CopyObjectCommand({
      Bucket: bucket,
      CopySource: `${bucket}/${encodedSource}`,
      Key: to,
    }),
  );
  await s3.send(new DeleteObjectCommand({ Bucket: bucket, Key: from }));
};

npx cdk deploy

facilityCode,facilityName,prefecture,isActive
TES01,テスト練習場01,東京都,true
TES02,テスト練習場02,神奈川県,true
TES03,テスト練習場03,大阪府,false

aws s3 cp facilities.csv s3://<作成されたバケット名>/uploads/facilities.csv

取り込み完了: 3 件

aws dynamodb scan --table-name facility