Migration

# Step 1: Export from Neo4j using neo4j-admin or APOC
# Nodes CSV format for Neptune:
# ~id, ~label, property1:String, property2:Int
# "person-1", "Person", "Alice", 30

# Edges CSV format for Neptune:
# ~id, ~from, ~to, ~label, property1:String
# "edge-1", "person-1", "person-2", "KNOWS", "2024-01-01"

// Export nodes
CALL apoc.export.csv.query(
  "MATCH (n) RETURN id(n) AS `~id`, labels(n)[0] AS `~label`, n.name AS `name:String`, n.age AS `age:Int`",
  "nodes.csv", {}
)

// Export relationships
CALL apoc.export.csv.query(
  "MATCH (a)-[r]->(b) RETURN id(r) AS `~id`, id(a) AS `~from`, id(b) AS `~to`, type(r) AS `~label`, r.since AS `since:String`",
  "edges.csv", {}
)

# Upload CSVs to S3
aws s3 cp nodes.csv s3://your-bucket/migration/
aws s3 cp edges.csv s3://your-bucket/migration/

# Start Neptune bulk loader
curl -X POST \
  "https://your-cluster:8182/loader" \
  -H "Content-Type: application/json" \
  -d '{
    "source": "s3://your-bucket/migration/",
    "format": "csv",
    "iamRoleArn": "arn:aws:iam::123456789012:role/NeptuneLoadFromS3",
    "region": "us-east-1",
    "failOnError": "FALSE",
    "parallelism": "MEDIUM"
  }'

# Check loader status
curl "https://your-cluster:8182/loader/{load-id}"

import boto3

analytics_client = boto3.client('neptune-graph')

# Load from S3 (after exporting from Neo4j to CSV/Parquet).
#
# roleArn: scope to LEAST PRIVILEGE — s3:GetObject and s3:ListBucket on the
# specific bucket/prefix only. Do NOT attach AmazonS3FullAccess or
# Resource:"*" to import roles.
analytics_client.create_graph_using_import_task(
    graphName='migrated-graph',
    source='s3://your-bucket/migration/',
    format='CSV',
    roleArn='arn:aws:iam::123456789012:role/NeptuneAnalyticsImportRole',
    deletionProtection=True,
    # Mandatory tags — a graph missing either tag is a failed task.
    tags={'created_by': 'neptune-skill', 'generation_model': '<model-id>'},
)

aws neptune-graph create-graph-using-import-task \
  --graph-name migrated-graph \
  --source 's3://your-bucket/migration/' \
  --format CSV \
  --role-arn 'arn:aws:iam::123456789012:role/NeptuneAnalyticsImportRole' \
  --deletion-protection \
  --tags created_by=neptune-skill,generation_model=<model-id>

class DualWriteClient:
    """Write to both Neo4j and Neptune during migration."""

    def __init__(self, neo4j_driver, neptune_client):
        self.neo4j = neo4j_driver
        self.neptune = neptune_client

    def create_node(self, label: str, properties: Dict):
        # Write to Neo4j (primary)
        self.neo4j.session().run(
            f"CREATE (n:{label} $props)", props=properties
        )
        # Write to Neptune (secondary)
        # Use Gremlin bindings (parameterized) — never f-string interpolate
        # label/property values into the query string (injection risk).
        query = "g.addV(label)"
        bindings: Dict = {"label": label}
        for i, (k, v) in enumerate(properties.items()):
            key_b, val_b = f"k{i}", f"v{i}"
            query += f".property({key_b}, {val_b})"
            bindings[key_b] = k
            bindings[val_b] = v
        self.neptune.submit(query, bindings=bindings).all().result()

-- Simple MATCH
MATCH (p:Person {name: 'Alice'})-[:KNOWS]->(friend)
RETURN friend.name

-- Filtering
MATCH (p:Person)
WHERE p.age > 25 AND p.city = 'Seattle'
RETURN p.name, p.age

-- Aggregation
MATCH (p:Person)-[:PURCHASED]->(product)
RETURN product.name, count(p) AS buyers
ORDER BY buyers DESC
LIMIT 10

-- Path patterns
MATCH path = (a:Person {name: 'Alice'})-[:KNOWS*1..3]->(b:Person {name: 'Bob'})
RETURN path

-- CALL { } subqueries — supported read-only (no rewrite needed)
-- Previously listed as unsupported. Older migration guides that rewrote
-- to OPTIONAL MATCH no longer apply.
MATCH (p:Person)
CALL {
  WITH p
  MATCH (p)-[:PURCHASED]->(product)
  RETURN count(product) AS purchaseCount
}
RETURN p.name, purchaseCount

// Neo4j APOC:
// CALL apoc.path.expandConfig(startNode, {maxLevel: 4, relationshipFilter: "KNOWS>"})

// Neptune Gremlin equivalent:
g.V().has('Person', 'name', 'Alice')
  .repeat(out('KNOWS').simplePath())
  .times(4)
  .dedup()
  .valueMap('name')

-- Neo4j (built-in full-text index)
CALL db.index.fulltext.queryNodes("personIndex", "Ali*")
YIELD node
RETURN node.name

-- Neptune: Use OpenSearch for full-text, then look up in Neptune
-- Step 1: Query OpenSearch for matching IDs
-- Step 2: MATCH (p:Person) WHERE p.id IN $matchedIds RETURN p

# Neo4j triggers fire on write events
# Neptune equivalent: enable Neptune Streams and process with Lambda

# Enable streams on cluster (one-time setup)
# aws neptune modify-db-cluster --enable-cloudwatch-logs-exports '["audit"]'

# Lambda processes stream events
def handle_neptune_stream(event, context):
    for record in event['records']:
        if record['eventName'] == 'INSERT' and record['data']['type'] == 'vl':
            # New vertex created — equivalent to Neo4j trigger
            vertex_label = record['data']['value']
            # ... trigger logic here