Neptune Bulk Load from S3 — Gremlin CSV Format, Loader API, and Streams CDC

When to use bulk load vs Gremlin insert

Gremlin CSV file format for vertices

Neptune's bulk loader expects separate CSV files for vertices and for edges. Vertex files must have a header row containing ~id and ~label plus any property columns. Property column names follow the format propertyName:datatype where datatype is one of String, Long, Double, Bool, Date.

# vertices.csv — entities for a knowledge graph

~id,~label,name:String,entityType:String,confidence:Double,createdAt:Long
"entity::aws::lambda","Entity","AWS Lambda","Service",0.99,1699430400000
"entity::aws::dynamodb","Entity","Amazon DynamoDB","Service",0.99,1699430400000
"concept::serverless","Concept","serverless computing","Concept",1.0,1699430400000
"concept::nosql","Concept","NoSQL database","Concept",1.0,1699430400000

Column type suffixes in the header tell Neptune's parser how to coerce values. Columns without a suffix default to String. Date columns must use ISO 8601 format (2026-11-08T00:00:00Z). Long columns must be integer milliseconds (epoch-ms) for epoch timestamps — do not use ISO 8601 for Long columns.

Cardinality suffixes: append (single) or (set) to the column name to control multi-value behavior: tags:String(set). Set cardinality allows multiple values for the same property key. The column value uses semicolons to separate multiple values: "tag1;tag2;tag3".

Gremlin CSV file format for edges

Edge files require four mandatory header columns: ~id (edge ID), ~from (source vertex ID), ~to (target vertex ID), and ~label (edge type). Both ~from and ~to must reference vertex IDs present in vertex files submitted in the same or prior load jobs. If a referenced vertex ID does not exist, the edge is silently skipped by default (use failOnError: true in the load request to fail instead).

# edges.csv — RELATES_TO edges

~id,~from,~to,~label,relationshipType:String,weight:Double,validFrom:Long
"edge::001","entity::aws::lambda","concept::serverless","RELATES_TO","uses",0.95,1699430400000
"edge::002","entity::aws::dynamodb","concept::nosql","RELATES_TO","implements",0.98,1699430400000
"edge::003","entity::aws::lambda","entity::aws::dynamodb","RELATES_TO","integrates_with",0.90,1699430400000

Neptune bulk loader processes vertex files and edge files in separate passes. Always submit vertex files first (or include both and let Neptune's two-pass loading handle ordering). If you submit only edge files and vertex IDs are missing, edges are dropped without error unless failOnError is enabled.

IAM prerequisites for the loader

Neptune's loader assumes an IAM role to read from S3. The role must trust the rds.amazonaws.com service (not neptune.amazonaws.com — a common misconfiguration). The Neptune cluster must also be configured with the role ARN.

// CDK: create the S3 access role for Neptune bulk load
const neptuneS3Role = new iam.Role(this, 'NeptuneS3LoadRole', {
  assumedBy: new iam.ServicePrincipal('rds.amazonaws.com'),  // rds, NOT neptune
  description: 'Neptune bulk loader S3 read access',
});

neptuneS3Role.addToPolicy(new iam.PolicyStatement({
  actions: [
    's3:GetObject',
    's3:ListBucket',
    's3:GetBucketLocation',
  ],
  resources: [
    graphDataBucket.bucketArn,
    graphDataBucket.arnForObjects('*'),
  ],
}));

// Associate the role with the Neptune cluster
// (CDK neptune alpha module supports associatedRoles)
const cluster = new neptune.DatabaseCluster(this, 'McpGraph', {
  // ...
  associatedRoles: [neptuneS3Role],
});

# CLI: associate role with existing cluster (if not done in CDK)
aws neptune add-role-to-db-cluster \
  --db-cluster-identifier your-neptune-cluster \
  --role-arn arn:aws:iam::ACCOUNT:role/NeptuneS3LoadRole

The S3 bucket must be in the same AWS region as the Neptune cluster. Cross-region bulk loads are not supported. If your data pipeline produces files in a different region, copy them to a same-region bucket before submitting the load job.

Submitting and monitoring a load job

The Neptune Loader API is an HTTP REST endpoint exposed on the Neptune cluster endpoint at port 8182. Submit a POST request (also signed with SigV4 if IAM database authentication is enabled).

import https from 'https';
import { SignatureV4 } from '@aws-sdk/signature-v4';
import { Sha256 } from '@aws-crypto/sha256-js';
import { defaultProvider } from '@aws-sdk/credential-provider-node';

const NEPTUNE_ENDPOINT = process.env.NEPTUNE_ENDPOINT!;
const REGION = process.env.AWS_REGION || 'us-east-1';

async function submitBulkLoad(s3Prefix: string, roleArn: string): Promise<string> {
  const body = JSON.stringify({
    source: s3Prefix,            // e.g. s3://your-bucket/graph-data/
    format: 'csv',               // csv for Gremlin CSV format
    iamRoleArn: roleArn,
    region: REGION,
    failOnError: 'FALSE',        // skip bad records; set TRUE to fail on first error
    parallelism: 'MEDIUM',       // LOW | MEDIUM | HIGH | OVERSUBSCRIBE
    updateSingleCardinalityProperties: 'TRUE',  // overwrite existing single-cardinality properties
    queueRequest: 'TRUE',        // queue if another load job is running
  });

  const response = await signedRequest('POST', `https://${NEPTUNE_ENDPOINT}:8182/loader`, body);
  return response.payload.loadId;
}

async function pollLoadStatus(loadId: string): Promise<void> {
  for (let attempt = 0; attempt < 120; attempt++) {
    await new Promise(r => setTimeout(r, 30_000));  // poll every 30 s

    const status = await signedRequest('GET', `https://${NEPTUNE_ENDPOINT}:8182/loader/${loadId}?details=TRUE`);
    const { overallStatus, errors } = status.payload;

    if (overallStatus.status === 'LOAD_COMPLETED') {
      console.log(`Load completed: ${overallStatus.totalRecords} records in ${overallStatus.elapsedSeconds}s`);
      return;
    }
    if (overallStatus.status.startsWith('LOAD_FAILED') || overallStatus.status === 'LOAD_CANCELLED_BY_USER') {
      console.error('Load failed:', errors);
      throw new Error(`Neptune load failed: ${overallStatus.status}`);
    }
    console.log(`Load in progress: ${overallStatus.status} — ${overallStatus.fullLoadProgressPercent}% complete`);
  }
  throw new Error('Load timed out after 60 minutes');
}

Load status values: LOAD_NOT_STARTED, LOAD_IN_QUEUE, LOAD_IN_PROGRESS, LOAD_COMPLETED, LOAD_COMPLETED_WITH_ERRORS, LOAD_FAILED, LOAD_CANCELLED_BY_USER, LOAD_UNEXPECTED_ERROR. LOAD_COMPLETED_WITH_ERRORS means some records were skipped — use details=TRUE&errors=TRUE in the GET request to retrieve the error details with Amazon S3 path, row number, and error type.

# List all recent load jobs
curl -s -k "https://${NEPTUNE_ENDPOINT}:8182/loader?limit=10&includeQueuedLoads=TRUE" | jq '.payload.loadIds'

# Cancel a running load
curl -s -k -X DELETE "https://${NEPTUNE_ENDPOINT}:8182/loader/${LOAD_ID}"

Neptune Streams for change data capture

Neptune Streams is a change log that records every graph mutation in order. After the bulk load, enable Streams to capture incremental changes for downstream consumers — syncing to OpenSearch, triggering EventBridge events on specific entity changes, or building a materialized summary table in DynamoDB.

Streams is not enabled by default. Enable it via cluster parameter: set neptune_streams to 1. Streams data is available via HTTP at GET /gremlin/stream. Each record contains a commit number, operation type (ADD, REMOVE), entity type (vl = vertex label, vp = vertex property, e = edge, ep = edge property), and the data.

// Lambda poller for Neptune Streams — reads changes and pushes to EventBridge
import { DynamoDBClient, GetItemCommand, PutItemCommand } from '@aws-sdk/client-dynamodb';
import { EventBridgeClient, PutEventsCommand } from '@aws-sdk/client-eventbridge';

const ddb = new DynamoDBClient({});
const eb = new EventBridgeClient({});
const CURSOR_TABLE = process.env.CURSOR_TABLE!;  // DynamoDB table for commit cursor
const NEPTUNE_STREAM_URL = `https://${process.env.NEPTUNE_ENDPOINT}:8182/gremlin/stream`;

async function getLastCommitNum(): Promise<number> {
  const result = await ddb.send(new GetItemCommand({
    TableName: CURSOR_TABLE,
    Key: { pk: { S: 'neptune-streams-cursor' } },
  }));
  return result.Item?.commitNum ? parseInt(result.Item.commitNum.N!) : 1;
}

async function saveCommitNum(commitNum: number) {
  await ddb.send(new PutItemCommand({
    TableName: CURSOR_TABLE,
    Item: { pk: { S: 'neptune-streams-cursor' }, commitNum: { N: commitNum.toString() } },
  }));
}

export async function handler() {
  const lastCommit = await getLastCommitNum();

  // Fetch up to 100 records after the last processed commit
  const response = await signedRequest(
    'GET',
    `${NEPTUNE_STREAM_URL}?commitNum=${lastCommit}&limit=100&iteratorType=AFTER_SEQUENCE_NUMBER`,
  );

  const { records, lastEventId } = response;
  if (!records || records.length === 0) return;

  // Process records — emit EventBridge events for entity additions
  const events = records
    .filter((r: any) => r.type === 'vl' && r.op === 'ADD')  // new vertex label = new vertex
    .map((r: any) => ({
      EventBusName: 'default',
      Source: 'alivemcp.neptune',
      DetailType: 'EntityCreated',
      Detail: JSON.stringify({
        entityId: r.data.id,
        label: r.data.value,
        commitNum: r.commitTimestamp,
      }),
    }));

  if (events.length > 0) {
    await eb.send(new PutEventsCommand({ Entries: events }));
  }

  // Advance cursor
  await saveCommitNum(lastEventId.commitNum);
}

Streams limitations: only available on Neptune DB (not Neptune Serverless), retention is 7 days by default, records are not ordered within a millisecond window (use commitNum + opNum for total ordering), and there is no cross-region replication via Streams. For multi-region replication, combine Streams with S3 export and cross-region copy.

Common failures