戯言と技術 | Floci × DynamoDB ハンズオン — トランザクション・Streams対応のローカル検証【2026年版】

この記事でわかること
Floci上でDynamoDBテーブルを作成・操作する基本
トランザクション（TransactWriteItems）・条件付き書き込み
DynamoDB Streamsのローカル検証方法
GSI / LSI の作成例
DynamoDB Local との違いと使い分け

Floci は DynamoDB をインプロセスで再現し、トランザクションとStreamsにも対応しています。AWS公式の DynamoDB Local と比較して、S3/Lambda/SQSなど他サービスと同一コンテナで動かせるのが最大の強みです。

Floci の基本は入門記事を参照してください。

セットアップ

services:
  floci:
    image: hectorvent/floci:latest
    ports:
      - "4566:4566"
    environment:
      - FLOCI_STORAGE_MODE=persistent
    volumes:
      - ./floci-data:/app/data

テーブル作成

AWS CLI

aws dynamodb create-table \
  --table-name Users \
  --attribute-definitions AttributeName=userId,AttributeType=S \
  --key-schema AttributeName=userId,KeyType=HASH \
  --billing-mode PAY_PER_REQUEST \
  --endpoint-url http://localhost:4566

Python（boto3）

import boto3
ddb = boto3.client(
    "dynamodb",
    endpoint_url="http://localhost:4566",
    region_name="us-east-1",
    aws_access_key_id="test",
    aws_secret_access_key="test",
)

ddb.create_table(
    TableName="Users",
    KeySchema=[{"AttributeName": "userId", "KeyType": "HASH"}],
    AttributeDefinitions=[{"AttributeName": "userId", "AttributeType": "S"}],
    BillingMode="PAY_PER_REQUEST",
)

CRUD操作

# Put
ddb.put_item(
    TableName="Users",
    Item={"userId": {"S": "u001"}, "name": {"S": "Alice"}, "age": {"N": "30"}},
)

# Get
res = ddb.get_item(TableName="Users", Key={"userId": {"S": "u001"}})
print(res["Item"])

# Update
ddb.update_item(
    TableName="Users",
    Key={"userId": {"S": "u001"}},
    UpdateExpression="SET age = :a",
    ExpressionAttributeValues={":a": {"N": "31"}},
)

# Delete
ddb.delete_item(TableName="Users", Key={"userId": {"S": "u001"}})

高レベルAPI（boto3 resource）

resource = boto3.resource("dynamodb", endpoint_url="http://localhost:4566",
                          region_name="us-east-1",
                          aws_access_key_id="test", aws_secret_access_key="test")
table = resource.Table("Users")
table.put_item(Item={"userId": "u002", "name": "Bob", "age": 25})

GSI / LSI の作成

ddb.create_table(
    TableName="Orders",
    KeySchema=[
        {"AttributeName": "orderId", "KeyType": "HASH"},
        {"AttributeName": "createdAt", "KeyType": "RANGE"},
    ],
    AttributeDefinitions=[
        {"AttributeName": "orderId", "AttributeType": "S"},
        {"AttributeName": "createdAt", "AttributeType": "S"},
        {"AttributeName": "userId", "AttributeType": "S"},
    ],
    GlobalSecondaryIndexes=[{
        "IndexName": "UserIndex",
        "KeySchema": [{"AttributeName": "userId", "KeyType": "HASH"}],
        "Projection": {"ProjectionType": "ALL"},
    }],
    BillingMode="PAY_PER_REQUEST",
)

GSI経由のクエリも本番と同じAPIで動作確認できます。

トランザクション（TransactWriteItems）

複数テーブル・複数アイテムの原子的更新もFloci上で検証可能:

ddb.transact_write_items(TransactItems=[
    {"Put": {
        "TableName": "Users",
        "Item": {"userId": {"S": "u100"}, "balance": {"N": "1000"}},
    }},
    {"Update": {
        "TableName": "Accounts",
        "Key": {"accountId": {"S": "a001"}},
        "UpdateExpression": "ADD deposits :v",
        "ExpressionAttributeValues": {":v": {"N": "1000"}},
    }},
])

条件付き書き込み

# 存在しないときだけPut（冪等性担保）
try:
    ddb.put_item(
        TableName="Users",
        Item={"userId": {"S": "u001"}, "name": {"S": "Alice"}},
        ConditionExpression="attribute_not_exists(userId)",
    )
except ddb.exceptions.ConditionalCheckFailedException:
    print("already exists")

DynamoDB Streams

Streams有効化

ddb.update_table(
    TableName="Users",
    StreamSpecification={
        "StreamEnabled": True,
        "StreamViewType": "NEW_AND_OLD_IMAGES",
    },
)

ストリームのポーリング

streams = boto3.client("dynamodbstreams", endpoint_url="http://localhost:4566",
                      region_name="us-east-1",
                      aws_access_key_id="test", aws_secret_access_key="test")

desc = ddb.describe_table(TableName="Users")["Table"]
stream_arn = desc["LatestStreamArn"]

shards = streams.describe_stream(StreamArn=stream_arn)["StreamDescription"]["Shards"]
shard_id = shards[0]["ShardId"]

it = streams.get_shard_iterator(
    StreamArn=stream_arn, ShardId=shard_id, ShardIteratorType="TRIM_HORIZON",
)["ShardIterator"]

records = streams.get_records(ShardIterator=it)["Records"]
for r in records:
    print(r["dynamodb"])

Lambdaと組み合わせると、Stream駆動の非同期処理までローカルで完結できます。

Node.js（AWS SDK v3）例

import { DynamoDBClient, PutItemCommand } from "@aws-sdk/client-dynamodb";

const client = new DynamoDBClient({
  endpoint: "http://localhost:4566",
  region: "us-east-1",
  credentials: { accessKeyId: "test", secretAccessKey: "test" },
});

await client.send(new PutItemCommand({
  TableName: "Users",
  Item: { userId: { S: "u001" }, name: { S: "Alice" } },
}));

DynamoDB Localとの違い

項目	Floci	DynamoDB Local
対応サービス	DynamoDB + S3/Lambda/SQSなど24種	DynamoDBのみ
起動時間	~24ms	~2秒（JVM起動）
イメージサイズ	~90MB	~150MB
Streams	対応	対応
トランザクション	対応	対応
S3イベント連携テスト	可能	不可（DynamoDB単独）

DynamoDBだけ検証したい → DynamoDB Local でも可。 他サービスと組み合わせたE2Eテスト → Floci が圧倒的に便利。

よくあるエラー

`ResourceNotFoundException`

テーブルが作成されていない。memory モードでの再起動後は特に発生しやすい。seedスクリプトで自動作成を。

`ValidationException: Missing key`

キースキーマに定義した属性が Item に存在しない。AttributeDefinitions と Item の整合を確認。

Streams が取得できない

StreamSpecification.StreamEnabled=True になっているか、describe_table でStreamArnを取得できるかを確認。

FAQ

Q. PartiQLは使える？

Flociは基本的なPartiQLクエリに対応しています（ExecuteStatement）。

Q. オンデマンドとプロビジョンドの違いはエミュレートされる？

両方受け付けますが、スロットリングや課金挙動は再現されません。負荷試験は実AWSで。

Q. バックアップ・リストアは？

CreateBackup / RestoreTableFromBackup は未対応。データ永続化は FLOCI_STORAGE_MODE=persistent で代替。

バッチ操作

BatchWriteItem

ddb.batch_write_item(RequestItems={
    "Users": [
        {"PutRequest": {"Item": {"userId": {"S": "u010"}, "name": {"S": "Charlie"}}}},
        {"PutRequest": {"Item": {"userId": {"S": "u011"}, "name": {"S": "Dana"}}}},
        {"DeleteRequest": {"Key": {"userId": {"S": "u001"}}}},
    ]
})

最大25件/1リクエスト。未処理項目（UnprocessedItems）が返る挙動もFloci上で再現されます。

BatchGetItem

res = ddb.batch_get_item(RequestItems={
    "Users": {"Keys": [{"userId": {"S": "u010"}}, {"userId": {"S": "u011"}}]}
})
print(res["Responses"]["Users"])

PartiQL クエリ

SQLライクな構文でDynamoDBを操作可能:

# SELECT
res = ddb.execute_statement(Statement="SELECT * FROM Users WHERE userId = 'u010'")
print(res["Items"])

# INSERT
ddb.execute_statement(
    Statement="INSERT INTO Users VALUE {'userId': 'u020', 'name': 'Eve'}"
)

# UPDATE
ddb.execute_statement(
    Statement="UPDATE Users SET age = 28 WHERE userId = 'u020'"
)

# DELETE
ddb.execute_statement(Statement="DELETE FROM Users WHERE userId = 'u020'")

既存RDB経験者の学習コストを下げる手段として有効です。

Single Table Design の検証

DynamoDB設計のデファクト「Single Table Design」もFlociで検証可能:

ddb.create_table(
    TableName="AppData",
    KeySchema=[
        {"AttributeName": "PK", "KeyType": "HASH"},
        {"AttributeName": "SK", "KeyType": "RANGE"},
    ],
    AttributeDefinitions=[
        {"AttributeName": "PK", "AttributeType": "S"},
        {"AttributeName": "SK", "AttributeType": "S"},
        {"AttributeName": "GSI1PK", "AttributeType": "S"},
        {"AttributeName": "GSI1SK", "AttributeType": "S"},
    ],
    GlobalSecondaryIndexes=[{
        "IndexName": "GSI1",
        "KeySchema": [
            {"AttributeName": "GSI1PK", "KeyType": "HASH"},
            {"AttributeName": "GSI1SK", "KeyType": "RANGE"},
        ],
        "Projection": {"ProjectionType": "ALL"},
    }],
    BillingMode="PAY_PER_REQUEST",
)

# User + Order を同一テーブルに格納
ddb.put_item(TableName="AppData", Item={
    "PK": {"S": "USER#u001"}, "SK": {"S": "PROFILE"},
    "name": {"S": "Alice"}, "GSI1PK": {"S": "USER"}, "GSI1SK": {"S": "Alice"},
})
ddb.put_item(TableName="AppData", Item={
    "PK": {"S": "USER#u001"}, "SK": {"S": "ORDER#o100"},
    "amount": {"N": "1500"}, "GSI1PK": {"S": "ORDER"}, "GSI1SK": {"S": "2026-04-15"},
})

アクセスパターン設計の妥当性を実APIで検証できるのは大きなメリットです。

DAX（キャッシュ）の代替

FlociはDAXを直接サポートしないため、ローカルでは以下で代替:

アプリ側にインメモリキャッシュ（Redis/ElastiCacheのFloci対応や、cachetools）
ヒット率の検証だけは DynamoDB単体で実施
本番相当のレイテンシ検証は実AWSで

Node.js（DocumentClient）例

import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
import { DynamoDBDocumentClient, PutCommand, QueryCommand }
  from "@aws-sdk/lib-dynamodb";

const base = new DynamoDBClient({
  endpoint: "http://localhost:4566",
  region: "us-east-1",
  credentials: { accessKeyId: "test", secretAccessKey: "test" },
});
const ddb = DynamoDBDocumentClient.from(base);

await ddb.send(new PutCommand({
  TableName: "AppData",
  Item: { PK: "USER#u010", SK: "PROFILE", name: "Fay" },
}));

const res = await ddb.send(new QueryCommand({
  TableName: "AppData",
  KeyConditionExpression: "PK = :pk",
  ExpressionAttributeValues: { ":pk": "USER#u010" },
}));

パフォーマンス検証シナリオ

Flociは課金やスロットリングを再現しませんが、クエリ効率のボトルネック検出は可能:

Scan vs Query の実行時間比較
GSI有無でのレイテンシ差
大量データ投入後の LastEvaluatedKey ページネーション挙動

import time
items = [{"PutRequest": {"Item": {"PK": {"S": f"K#{i:05}"}, "SK": {"S": "#"}}}}
         for i in range(1000)]
for i in range(0, 1000, 25):
    ddb.batch_write_item(RequestItems={"AppData": items[i:i+25]})

t = time.time()
ddb.scan(TableName="AppData")
print(f"scan: {time.time()-t:.3f}s")

本番DynamoDBとの差異

プロビジョンドキャパシティのスロットリングは再現されない
グローバルテーブル（マルチリージョン）未対応
バックアップ / PITR 未対応
DAX 未対応
課金・パーティション分割は考慮されない

まとめ

FlociはDynamoDBのCRUD・GSI・トランザクション・Streamsをローカル再現
S3イベントやLambdaトリガーと統合E2Eテストが可能
DynamoDB Localより周辺サービス込みで便利
スロットリングや課金挙動は再現されないため、本番相当検証は別途

Floci × DynamoDB ハンズオン — トランザクション・Streams対応のローカル検証【2026年版】