DEV Community

Cover image for DynamoDB Bulk Put Items: Load S3 Data into Existing Tables
Sascha Mirow-Stanitzok
Sascha Mirow-Stanitzok

Posted on Edited on Originally published at sascha-ms.de

DynamoDB Bulk Put Items: Load S3 Data into Existing Tables

#aws #database #serverless #tutorial

With DynamoDB bulk put items you load large amounts of data from S3 straight into a table that already exists and is serving production traffic. Until now that was a gap: import from S3 always created a new table, and for everything else you built your own pipeline with Lambda, Step Functions or Glue, raised capacity and hoped the application would not get throttled in the meantime.

Since November 5, 2026, DynamoDB handles this part itself. What interests me most is not the speed but a small detail: every record in the input file can carry its own condition. That makes a nightly sync safe to repeat without overwriting newer data written by the application. In this article you build the whole thing once, end to end, with the Serverless Framework, TypeScript and a report that you actually evaluate at the end.

In short

Bulk put items writes data from S3 into an existing DynamoDB table without consuming the table’s read or write capacity. The key ingredient is a condition per record, which makes an import safe to repeat. The biggest pitfall: cancelled runs are not rolled back, and every written record also shows up in your streams.

What you will learn in this article

  • When bulk put items fits better than BatchWriteItem or importing into a new table
  • How to generate input files with a version guard per record in TypeScript
  • How to start and monitor the import via CLI, SDK and a nightly Lambda function
  • How to read the error report and manifest, and what the whole thing costs

What is DynamoDB bulk put items?

Bulk put items is a managed DynamoDB feature. You point it at a folder (more precisely, a prefix) in an S3 bucket, name the target table and a second S3 location for the report. The service does the rest: it reads the files, writes every item, retries failed writes and drops a report at the end.

Two properties make the difference. First, the writes take a separate path and consume no read or write capacity of the table. Your application does not notice, even on a table in provisioned mode. Second, new items are readable right away while the import is still running. There is no switch-over moment at the end.

The API consists of four calls: StartBulkPutItems starts an import, DescribeBulkOperation returns its progress, ListBulkOperations shows the runs of the last 90 days and CancelBulkOperation stops a run. The start call returns an ARN right away, a unique identifier for the operation. After that the actual import runs asynchronously.

DynamoDB accepts two input formats: DynamoDB JSON as JSON Lines (one JSON object per line) or Amazon Ion. Both can be uncompressed, gzip-compressed or Zstandard-compressed. All files under one prefix must share the same format and compression.

Bulk put items, BatchWriteItem or import from S3?

DynamoDB now offers three ways to get many items into a table. They compete less than it seems at first glance. The table below shows which one I would pick for what.

BatchWriteItem / PutItem Import from S3 (ImportTable) Bulk put items
Target table existing created new existing
Typical volume a few to thousands of items large initial load thousands to billions of items
Consumes table capacity yes no no
Condition per item PutItem only, not BatchWriteItem no yes
Latency seconds minutes to hours minutes to hours
Global tables yes no no (as of November 2026)

For a few hundred records from a Lambda function, BatchWriteItem stays the right choice. AWS says so itself, and it matches my view: an asynchronous job with a report in S3 is simply too much ceremony for small volumes. Bulk put items shines for syncs, backfills and migrations. In other words, everywhere you would previously have built your own pipeline.

If you are still deciding which service fits your use case, the AWS service finder may help you sort it out.

How an import runs

🎬 Flow of a DynamoDB bulk put items import – this animated diagram only runs on my site: watch it on sascha-ms.de

The diagram shows the example we are about to build. A script converts the supplier file into the input format and stores it in S3. Then DynamoDB takes over. Every successful write also creates a record in DynamoDB Streams if the table has streams enabled. That matters for everything attached to your streams, more on that later.

The example: a spare parts catalog for a bike rental

Picture a bike rental with its own workshop. A wholesaler delivers a CSV file every night with all spare parts: part number, name, price in cents and stock level. The workshop app reads from that table and also writes to it, for example when someone corrects a price by hand and bumps the catalog version.

This is where it gets interesting. The nightly import should create new parts and update older entries. It must not overwrite anything that already has a newer version, though. And if the run starts twice for whatever reason, nothing should break. Without a condition per record you would have to solve that in your own pipeline. With bulk put items it is one line in the input file.

Step by step: DynamoDB bulk put items with a version guard

Prerequisites

  • An AWS account and a Region with bulk put items (the Standard tier is available in all commercial Regions)
  • Node.js 22 or later and the Serverless Framework v4
  • An up-to-date AWS CLI v2 (older versions do not know the new commands yet)
  • @aws-sdk/client-dynamodb 3.1162.0 or later, I tested with 3.1163.0

Create a new project and install the dependencies:

mkdir bike-parts-catalog && cd bike-parts-catalog
npm init -y
npm pkg set type=module
npm install @aws-sdk/client-dynamodb @aws-sdk/client-s3 @aws-sdk/util-dynamodb
npm install -D typescript tsx @types/node @types/aws-lambda
Enter fullscreen mode Exit fullscreen mode 
// tsconfig.json
{
  "compilerOptions": {
    "target": "ES2022",
    "module": "NodeNext",
    "moduleResolution": "NodeNext",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "noEmit": true,
    "types": ["node"]
  },
  "include": ["src", "test"]
}
Enter fullscreen mode Exit fullscreen mode

Step 1: Table and buckets with the Serverless Framework

The infrastructure consists of a table with the part number sku as partition key, a bucket for the input files and a bucket for the reports. On top comes a Lambda function that starts the import every night. The IAM permissions are deliberately tight: dynamodb:StartBulkOperation applies account-wide, while dynamodb:BulkPutItems is scoped to the table.

# serverless.yml
service: bike-parts-catalog

build:
  esbuild:
    # Bundle the SDK: the bulk commands need @aws-sdk/client-dynamodb 3.1162.0 or later
    exclude: []

provider:
  name: aws
  runtime: nodejs24.x
  region: eu-central-1
  architecture: arm64
  environment:
    TABLE_NAME: !Ref PartsTable
    INPUT_BUCKET: !Ref InputBucket
    REPORT_BUCKET: !Ref ReportBucket
  iam:
    role:
      statements:
        - Effect: Allow
          Action:
            - dynamodb:StartBulkOperation
            - dynamodb:DescribeBulkOperation
            - dynamodb:ListBulkOperations
          Resource: "*"
        - Effect: Allow
          Action:
            - dynamodb:BulkPutItems
          Resource: !GetAtt PartsTable.Arn
        - Effect: Allow
          Action:
            - s3:ListBucket
          Resource: !GetAtt InputBucket.Arn
        - Effect: Allow
          Action:
            - s3:GetObject
          Resource: !Sub "${InputBucket.Arn}/incoming/*"
        - Effect: Allow
          Action:
            - s3:PutObject
            - s3:AbortMultipartUpload
          Resource: !Sub "${ReportBucket.Arn}/reports/*"

functions:
  startNightlyImport:
    handler: src/handler.startNightlyImport
    timeout: 30
    events:
      - schedule:
          method: scheduler
          rate: cron(15 2 * * ? *)
          timezone: Europe/Berlin

resources:
  Resources:
    PartsTable:
      Type: AWS::DynamoDB::Table
      Properties:
        TableName: bike-parts-catalog-${sls:stage}
        BillingMode: PAY_PER_REQUEST
        AttributeDefinitions:
          - AttributeName: sku
            AttributeType: S
        KeySchema:
          - AttributeName: sku
            KeyType: HASH
        PointInTimeRecoverySpecification:
          PointInTimeRecoveryEnabled: true
    InputBucket:
      Type: AWS::S3::Bucket
      Properties:
        LifecycleConfiguration:
          Rules:
            - Id: expire-incoming
              Status: Enabled
              Prefix: incoming/
              ExpirationInDays: 14
    ReportBucket:
      Type: AWS::S3::Bucket
      Properties:
        LifecycleConfiguration:
          Rules:
            - Id: expire-reports
              Status: Enabled
              Prefix: reports/
              ExpirationInDays: 30
  Outputs:
    InputBucketName:
      Value: !Ref InputBucket
    ReportBucketName:
      Value: !Ref ReportBucket
Enter fullscreen mode Exit fullscreen mode

Two details in this file deserve a second look. The build.esbuild.exclude: [] section makes sure the SDK gets bundled. Otherwise the Serverless Framework leaves out @aws-sdk/* and relies on the SDK in the runtime, which may not know the new command yet. In addition, lifecycle rules clean up old input files and reports automatically so nothing piles up in S3.

npx serverless deploy --stage dev
Enter fullscreen mode Exit fullscreen mode

Expected result: After the deployment the framework shows the names of both buckets under “Stack Outputs”. Note them down, you will need them shortly.

Step 2: Generate input files with a condition

Each line of the input file contains an Item field in DynamoDB JSON and, optionally, a ConditionExpression with placeholders. The syntax is exactly that of PutItem. The script reads the CSV file, sets the catalog version as a number (here the date, for example 20261106) and attaches the same guard to every record:

// src/build-input.ts
import { readFileSync, writeFileSync, mkdirSync } from "node:fs";
import { gzipSync } from "node:zlib";
import { marshall } from "@aws-sdk/util-dynamodb";

export interface PartRow {
  sku: string;
  name: string;
  priceCents: number;
  stock: number;
}

// One record in the bulk put items input format (one JSON Lines row)
export interface BulkRecord {
  Item: Record<string, unknown>;
  ConditionExpression?: string;
  ExpressionAttributeNames?: Record<string, string>;
  ExpressionAttributeValues?: Record<string, unknown>;
}

// Simple CSV parser for "sku;name;priceCents;stock" without quotes
export function parseCsv(csv: string): PartRow[] {
  const lines = csv.split(/\r?\n/).filter((l) => l.trim() !== "");
  const [, ...rows] = lines; // skip the header row
  return rows.map((line, i) => {
    const [sku, name, price, stock] = line.split(";");
    const priceCents = Number(price);
    const stockCount = Number(stock);
    if (!sku || !name || !Number.isInteger(priceCents) || !Number.isInteger(stockCount)) {
      throw new Error(`Line ${i + 2} is invalid: ${line}`);
    }
    return { sku: sku.trim(), name: name.trim(), priceCents, stock: stockCount };
  });
}

// Writes only if the part does not exist yet or our version is newer
export function toBulkRecord(row: PartRow, catalogVersion: number): BulkRecord {
  return {
    Item: marshall({
      sku: row.sku,
      name: row.name,
      priceCents: row.priceCents,
      stock: row.stock,
      catalogVersion,
    }),
    ConditionExpression: "attribute_not_exists(#v) OR #v < :v",
    ExpressionAttributeNames: { "#v": "catalogVersion" },
    ExpressionAttributeValues: marshall({ ":v": catalogVersion }),
  };
}

// Splits the records into similarly sized files and compresses them with gzip
export function toChunks(records: BulkRecord[], perFile: number): Buffer[] {
  const files: Buffer[] = [];
  for (let i = 0; i < records.length; i += perFile) {
    const body = records
      .slice(i, i + perFile)
      .map((r) => JSON.stringify(r))
      .join("\n");
    files.push(gzipSync(body + "\n"));
  }
  return files;
}

// Usage: npx tsx src/build-input.ts parts.csv 20261106 out/
if (process.argv[1]?.endsWith("build-input.ts")) {
  const [csvPath, version, outDir = "out"] = process.argv.slice(2);
  if (!csvPath || !version) {
    console.error("Usage: build-input.ts <csv> <catalog-version> [output-dir]");
    process.exit(1);
  }
  const rows = parseCsv(readFileSync(csvPath, "utf8"));
  const records = rows.map((r) => toBulkRecord(r, Number(version)));
  mkdirSync(outDir, { recursive: true });
  toChunks(records, 50_000).forEach((buf, i) => {
    writeFileSync(`${outDir}/parts-${String(i).padStart(4, "0")}.json.gz`, buf);
  });
  console.log(`Wrote ${records.length} records to ${outDir}/`);
}
Enter fullscreen mode Exit fullscreen mode

A finished line then looks like this:

{"Item":{"sku":{"S":"KET-108"},"name":{"S":"Chain 108 links"},"priceCents":{"N":"1999"},"stock":{"N":"40"},"catalogVersion":{"N":"20261106"}},"ConditionExpression":"attribute_not_exists(#v) OR #v < :v","ExpressionAttributeNames":{"#v":"catalogVersion"},"ExpressionAttributeValues":{":v":{"N":"20261106"}}}
Enter fullscreen mode Exit fullscreen mode

It is important that a record contains no line breaks. JSON.stringify without indentation takes care of that reliably. Splitting into files of 50,000 records each follows an AWS recommendation: several similarly sized files can be processed in parallel better than one huge file.

To make sure you can rely on the conversion, here are a few tests using the test runner built into Node.js:

// test/build-input.test.ts
import { test } from "node:test";
import assert from "node:assert/strict";
import { gunzipSync } from "node:zlib";
import { parseCsv, toBulkRecord, toChunks } from "../src/build-input.js";

const csv = "sku;name;priceCents;stock\nKET-108;Chain 108 links;1999;40\nSCH-28;Tube 28 inch;699;120\n";

test("parseCsv reads rows and skips the header", () => {
  const rows = parseCsv(csv);
  assert.equal(rows.length, 2);
  assert.deepEqual(rows[0], { sku: "KET-108", name: "Chain 108 links", priceCents: 1999, stock: 40 });
});

test("parseCsv reports invalid rows with their line number", () => {
  assert.throws(() => parseCsv("sku;name;priceCents;stock\nX;Y;abc;1"), /Line 2/);
});

test("toBulkRecord sets the version guard", () => {
  const rec = toBulkRecord(parseCsv(csv)[0]!, 20261106);
  assert.equal(rec.ConditionExpression, "attribute_not_exists(#v) OR #v < :v");
  assert.deepEqual(rec.ExpressionAttributeValues, { ":v": { N: "20261106" } });
  assert.deepEqual(rec.Item["sku"], { S: "KET-108" });
});

test("toChunks writes JSON Lines without line breaks inside a record", () => {
  const recs = parseCsv(csv).map((r) => toBulkRecord(r, 1));
  const files = toChunks(recs, 1);
  assert.equal(files.length, 2);
  const lines = gunzipSync(files[0]!).toString("utf8").trim().split("\n");
  assert.equal(lines.length, 1);
  assert.ok(JSON.parse(lines[0]!).Item);
});
Enter fullscreen mode Exit fullscreen mode 
npx tsc --noEmit
node --import tsx --test test/*.test.ts
npx tsx src/build-input.ts parts.csv 20261106 out
Enter fullscreen mode Exit fullscreen mode

Expected result: Four green tests and one or more files such as parts-0000.json.gz in the out/ folder.

Step 3: Upload the files to S3

Upload the files under a prefix that contains the catalog version. That keeps every run cleanly separated, and an import never accidentally reads the previous day’s files.

aws s3 cp out/ s3://<input-bucket>/incoming/20261106/ --recursive
Enter fullscreen mode Exit fullscreen mode

Step 4: Start the import with the AWS CLI

For a first test the CLI is enough. The output configuration requests both report types: ERROR for broken records and CONDITION_CHECK_FAILURE for records that failed their condition. Without the second one you learn nothing about skipped records, not even how many there were.

aws dynamodb start-bulk-put-items \
  --table-name bike-parts-catalog-dev \
  --client-token parts-20261106 \
  --input-s3-configuration '{"S3Bucket":"<input-bucket>","S3KeyPrefix":"incoming/20261106/","InputFormat":"DYNAMODB_JSON","InputCompressionType":"GZIP"}' \
  --output-s3-configuration '{"S3Bucket":"<report-bucket>","S3KeyPrefix":"reports/20261106","OutputType":["ERROR","CONDITION_CHECK_FAILURE"]}'

aws dynamodb describe-bulk-operation --bulk-operation-arn <arn-from-the-response>
Enter fullscreen mode Exit fullscreen mode

The client-token makes the start idempotent. If you call the same command again within eight hours with the same parameters, you get the same operation back instead of a second import. With different parameters and the same token, DynamoDB responds with IdempotentParameterMismatchException.

Expected result: At first the status is IN_PROGRESS, and the counters stay at zero for a few minutes while DynamoDB prepares the operation. After that they rise in steps because the values are only refreshed periodically.

Step 5: Start and monitor the import from TypeScript

Day to day you do not want to type this by hand. The following module wraps starting and monitoring. waitForImport checks the status every 30 seconds and stops at one of the four final states.

// src/bulk-import.ts
import {
  DynamoDBClient,
  StartBulkPutItemsCommand,
  DescribeBulkOperationCommand,
  type BulkOperationDescription,
} from "@aws-sdk/client-dynamodb";

const ddb = new DynamoDBClient({});

export interface ImportOptions {
  tableName: string;
  inputBucket: string;
  inputPrefix: string;
  reportBucket: string;
  reportPrefix: string;
  // Same token = same import, even if the start is called twice
  clientToken: string;
  maxWriteRequestUnits?: number;
}

export async function startImport(o: ImportOptions): Promise<string> {
  const res = await ddb.send(
    new StartBulkPutItemsCommand({
      TableName: o.tableName,
      ClientToken: o.clientToken,
      PerformanceTier: "STANDARD",
      MaxWriteRequestUnits: o.maxWriteRequestUnits,
      InputS3Configuration: {
        S3Bucket: o.inputBucket,
        S3KeyPrefix: o.inputPrefix,
        InputFormat: "DYNAMODB_JSON",
        InputCompressionType: "GZIP",
      },
      OutputS3Configuration: {
        S3Bucket: o.reportBucket,
        S3KeyPrefix: o.reportPrefix,
        OutputType: ["ERROR", "CONDITION_CHECK_FAILURE"],
      },
    }),
  );
  const arn = res.BulkOperationDescription?.BulkOperationArn;
  if (!arn) throw new Error("No BulkOperationArn in the response");
  return arn;
}

const FINAL = new Set(["COMPLETED", "COMPLETED_WITH_ERRORS", "FAILED", "CANCELLED"]);

export async function waitForImport(
  arn: string,
  pollSeconds = 30,
): Promise<BulkOperationDescription> {
  for (;;) {
    const { BulkOperationDescription: d } = await ddb.send(
      new DescribeBulkOperationCommand({ BulkOperationArn: arn }),
    );
    if (!d) throw new Error("Bulk operation not found");
    const pct = d.TotalInputBytes
      ? Math.round(((d.ProcessedInputBytes ?? 0) / d.TotalInputBytes) * 100)
      : 0;
    console.log(
      `${d.BulkOperationStatus}: ${d.ProcessedItemCount ?? 0} records, ` +
        `${d.FailedItemCount ?? 0} failed, ${pct} % of input read`,
    );
    if (FINAL.has(d.BulkOperationStatus ?? "")) return d;
    await new Promise((r) => setTimeout(r, pollSeconds * 1000));
  }
}
Enter fullscreen mode Exit fullscreen mode 
// src/start-import.ts
import { startImport, waitForImport } from "./bulk-import.js";

// Usage: npx tsx src/start-import.ts 20261106
const version = process.argv[2];
if (!version) {
  console.error("Usage: start-import.ts <catalog-version>");
  process.exit(1);
}

const arn = await startImport({
  tableName: process.env.TABLE_NAME ?? "bike-parts-catalog-dev",
  inputBucket: process.env.INPUT_BUCKET ?? "",
  inputPrefix: `incoming/${version}/`,
  reportBucket: process.env.REPORT_BUCKET ?? "",
  reportPrefix: `reports/${version}`,
  clientToken: `parts-${version}`,
});
console.log(`Started: ${arn}`);

const result = await waitForImport(arn);
if (result.BulkOperationStatus !== "COMPLETED") {
  console.error(`Final status ${result.BulkOperationStatus}: ${result.FailureCode ?? ""} ${result.FailureMessage ?? ""}`);
  process.exit(2);
}
Enter fullscreen mode Exit fullscreen mode 
export INPUT_BUCKET=<input-bucket> REPORT_BUCKET=<report-bucket>
npx tsx src/start-import.ts 20261106
Enter fullscreen mode Exit fullscreen mode

The four final states mean the following: COMPLETED says every record was written or deliberately skipped. With COMPLETED_WITH_ERRORS some records failed, usually because of broken data. FAILED points to a systemic problem such as missing permissions, and CANCELLED means someone stopped the run.

Step 6: Nightly start via Lambda

For regular operation a small function starts the import every night at 2:15 a.m. Berlin time. It deliberately does not wait for the end, because a large import can take much longer than the 15-minute maximum runtime of a Lambda function.

// src/handler.ts
import { startImport } from "./bulk-import.js";

// Starts the nightly import of the catalog the supplier has dropped off by then
export const startNightlyImport = async (): Promise<{ arn: string }> => {
  const version = new Date().toISOString().slice(0, 10).replaceAll("-", "");
  const arn = await startImport({
    tableName: process.env.TABLE_NAME!,
    inputBucket: process.env.INPUT_BUCKET!,
    inputPrefix: `incoming/${version}/`,
    reportBucket: process.env.REPORT_BUCKET!,
    reportPrefix: `reports/${version}`,
    clientToken: `parts-${version}`,
    // Throttles the import so stream consumers can keep up
    maxWriteRequestUnits: 2000,
  });
  console.log(JSON.stringify({ message: "Bulk import started", arn, version }));
  return { arn };
};
Enter fullscreen mode Exit fullscreen mode

With maxWriteRequestUnits you set an upper bound for the write rate; the smallest allowed value is 1024. This does not change your table’s capacity, it only slows down the import itself. Do not set it too low, though: every operation has a time limit, and an import that is throttled too hard can fail because of it.

Step 7: Read the report and the manifest

At the end, your report prefix contains a manifest and, if needed, gzip-compressed report files. The manifest manifest-files.json lists the status of each input file: FULLY_PROCESSED, PARTIALLY_PROCESSED, NOT_PROCESSED or NOT_AVAILABLE. The reports are split into the folders errors/ and condition-check-failures/.

// src/report.ts
import { S3Client, ListObjectsV2Command, GetObjectCommand } from "@aws-sdk/client-s3";
import { gunzipSync } from "node:zlib";

const s3 = new S3Client({});

interface ReportLine {
  itemLocation?: { inputSourceArn?: string; itemIndex?: number };
  errorMessages?: string[];
}

async function readObject(bucket: string, key: string): Promise<Buffer> {
  const res = await s3.send(new GetObjectCommand({ Bucket: bucket, Key: key }));
  return Buffer.from(await res.Body!.transformToByteArray());
}

// Summarizes the manifest, errors and skipped records of a bulk operation
export async function summarize(bucket: string, prefix: string) {
  const keys: string[] = [];
  let token: string | undefined;
  do {
    const page = await s3.send(
      new ListObjectsV2Command({ Bucket: bucket, Prefix: prefix, ContinuationToken: token }),
    );
    keys.push(...(page.Contents ?? []).map((o) => o.Key!));
    token = page.NextContinuationToken;
  } while (token);

  const summary = { incompleteFiles: [] as string[], errors: 0, skipped: 0, samples: [] as string[] };

  for (const key of keys) {
    if (key.endsWith("manifest-files.json")) {
      const lines = (await readObject(bucket, key)).toString("utf8").split("\n").filter(Boolean);
      for (const l of lines) {
        const m = JSON.parse(l) as { inputSourceArn: string; fileProcessingStatus: string };
        if (m.fileProcessingStatus !== "FULLY_PROCESSED") summary.incompleteFiles.push(m.inputSourceArn);
      }
    } else if (key.endsWith(".json.gz")) {
      const lines = gunzipSync(await readObject(bucket, key)).toString("utf8").split("\n").filter(Boolean);
      const isSkip = key.includes("/condition-check-failures/");
      for (const l of lines) {
        const r = JSON.parse(l) as ReportLine;
        if (isSkip) summary.skipped++;
        else {
          summary.errors++;
          if (summary.samples.length < 5) {
            summary.samples.push(`${r.itemLocation?.itemIndex ?? "?"}: ${(r.errorMessages ?? []).join("; ")}`);
          }
        }
      }
    }
  }
  return summary;
}

// Usage: npx tsx src/report.ts <report-bucket> reports/20261106
if (process.argv[1]?.endsWith("report.ts")) {
  const [bucket, prefix] = process.argv.slice(2);
  console.log(JSON.stringify(await summarize(bucket!, prefix!), null, 2));
}
Enter fullscreen mode Exit fullscreen mode 
npx tsx src/report.ts <report-bucket> reports/20261106
Enter fullscreen mode Exit fullscreen mode

By the way, the FailedItemCount from DescribeBulkOperation only includes validation errors. You count skipped records exclusively via the report files. For that, the role doing the evaluation needs its own read permissions (s3:ListBucket and s3:GetObject) on the report bucket. The import’s write permissions are not enough.

Conditions per record in DynamoDB bulk put items

Decision tree of the version guard in DynamoDB bulk put items: write new parts, overwrite older versions, otherwise skip

The version guard decides per record: write, overwrite or skip.

The condition attribute_not_exists(#v) OR #v < :v covers three cases. If the part does not exist yet, the attribute does not exist either, and DynamoDB writes. If the stored version is lower, the import wins. If the workshop app has meanwhile written an equal or higher version, the record is skipped and shows up in the conflict report, with old and new state side by side.

A skipped record does not count as an error. The run still ends with COMPLETED. That is intended, but it can surprise you if you only look at the status. You can also mix conditions in one file: create new parts only (attribute_not_exists(sku)), update prices with a version guard and overwrite master data without any condition.

There is one more reason for the version guard, slightly hidden in the documentation. After an internal failover, DynamoDB may repeat a write. Without a condition, an older state could overwrite a newer one that way. With a condition, the repeat ends up in the conflict report at most. If a record shows up there whose old and new state are identical, that is exactly what happened, and you can ignore it. You may know the principle from my article on idempotency in AWS Lambda.

How much does DynamoDB bulk put items cost?

You pay per gigabyte written. In US East (N. Virginia) the Standard tier costs 0.10 US dollars per GB and the Accelerated tier 0.20 US dollars. Every item counts as at least 1 KB; above that, metering is exact and not rounded up to full kilobytes. For other Regions check the pricing page.

Let us run the numbers for our example: 200,000 spare parts of around 600 bytes each are billed at the 1 KB minimum, so roughly 0.2 GB. In US East that comes to about two cents per run on the Standard tier. Three things push the cost up, though:

  • Each global or local secondary index an item is written to is billed separately.
  • Records that fail their condition still cost money, even if the part does not exist at all.
  • Report files in S3 cost regular storage, hence the lifecycle rule from step 1.

Records that already fail validation, such as broken JSON, are not billed. According to AWS, the Accelerated tier only pays off from about 1 GB and mainly for fixed time windows. Under good conditions AWS quotes around 100 million 1 KB items in under an hour for Accelerated.

Common mistakes and how to avoid them

Cancelling does not roll anything back

CancelBulkOperation stops the import, but items already written stay in the table. You end up with a partially updated state. With a version guard that is no drama, because you can simply repeat the same import later with a new token. Without a condition you need to know exactly what has already been written. The manifest helps with processedRanges per file.

Stream consumers get flooded

Every written record creates an entry in DynamoDB Streams and, if enabled, in Kinesis Data Streams. If your stream feeds a Lambda function that, say, sends emails or forwards events, it suddenly receives 200,000 events at once. Stream records are kept for 24 hours; a consumer that falls further behind loses data. So check beforehand who is attached to your stream, and cap the rate with MaxWriteRequestUnits. If you use streams for the outbox pattern, this applies twice.

Whole items only, no partial updates

Bulk put items always replaces the entire item. If the workshop app maintains an attribute that is missing from the supplier file, such as a storage location, it is gone after the import. Either include such attributes in the input, or move them into a separate item. For real partial updates, UpdateItem remains the tool.

More pitfalls

  • Only one import runs per table at a time, and not while an index is being created.
  • Global tables are not supported at launch; there you stick with BatchWriteItem.
  • DAX does not see these writes: cached items stay stale until their TTL expires.
  • The import does not show up in the CloudWatch metrics for consumed capacity. Monitor the operation status, not the table.
  • For a bucket in another account, S3BucketOwner must be correct. Otherwise the import starts, writes items and fails only at the end without a report.

My take

I consider bulk put items one of the more useful DynamoDB additions in recent years, precisely because it is so unspectacular. I would rather not know how many Lambda functions with batching logic, backoff and capacity math exist only to “load this file into the table”. That code can go now. In a Serverless architecture this is exactly the direction I like: fewer components of your own, more responsibility on the service.

Two things I view critically. First, the report does not take care of itself. If you only check for COMPLETED, you miss silently skipped records. Plan the evaluation from the start, for example as a second step that raises an alarm after the import when the number of conflicts suddenly rises. Second, not having to think about capacity tempts you to forget streams and downstream systems. The table does not notice the import, but your stream consumer certainly does.

FAQ

What is the difference between bulk put items and import from S3?

Import from S3 (ImportTable) always creates a new table. Bulk put items writes into an existing table while the application keeps running.

Which file formats does DynamoDB bulk put items support?

DynamoDB JSON as JSON Lines and Amazon Ion, each uncompressed, with gzip or with Zstandard. All files under one prefix must be structured the same way.

Can I update individual attributes with bulk put items?

No. It always writes whole items. For changes to individual attributes you keep using UpdateItem.

Do imported items trigger DynamoDB Streams?

Yes, every successful write creates a stream record. Skipped records do not. You cap the rate with MaxWriteRequestUnits.

Your next step

Find the Lambda function or script in your project that writes data to DynamoDB in batches today. Then check whether there is a version number or timestamp you can use to express a condition per record. If so, switching over is usually an afternoon’s work. If your table has items with an expiry date, also take a look at my article on DynamoDB TTL, because imported items expire just like all others.

I would like to know: which import would you move to bulk put items first, and what has held you back so far?

Sources

This post was created with AI assistance.

👉 Next step: Not sure which AWS service fits your use case? My AWS service finder asks a few questions and points you to the right one.

📬 Stay updated: New articles on serverless, AI and leading tech teams via RSS, browser push or the weekly newsletter (in German).

Originally published on sascha-ms.de. More articles on serverless, AI in everyday development and leading tech teams: sascha-ms.de/en

Top comments (0)