Replication, Partitioning, and Sharding Questions

Situation summary: you’re comparing how DynamoDB’s Global Secondary Indexes (GSIs) and Cassandra “global index” approaches maintain indexes, what consistency they provide, the write/amplification cost, and the practical limits when you need strongly consistent queries across shards/partitions.

High-level difference

• DynamoDB GSI: a first-class, managed secondary index maintained by DynamoDB. Updates are propagated by the service as part of the write pipeline but propagation to the GSI is effectively asynchronous; reads from GSIs are eventually consistent (DynamoDB does not support strongly consistent queries on GSIs).
• Cassandra “global index” approaches: there is no single built-in strongly-consistent global index. Common patterns are (a) materialized views (async and known consistency pitfalls), (b) a dedicated index table (denormalized) maintained by application logic, and (c) using lightweight transactions (LWT/Paxos) to get linearizable updates to an index table. Each has different trade-offs.

How each system maintains the index

• DynamoDB GSI:
  • Indexed attributes are copied to a GSI storage partition by DynamoDB’s internal update path.
  • Propagation is not synchronous: base-table write returns success while the GSI update is queued and applied; DynamoDB handles retry/repair.
  • Managed sharding and rebalancing are transparent to you.
• Cassandra global-index approaches:
  • Materialized views: Cassandra nodes asynchronously update the view when base data changes; updates are local but rely on anti-entropy/repair for correctness.
  • Dedicated index table: application writes (or lightweight transactions) write both base row and a row in the index table — i.e., manual denormalization.
  • LWT/Paxos: uses multi-round consensus to serialize updates for a partition key, giving linearizability for that partition.

Consistency guarantees

• DynamoDB GSI:
  • Base table: optionally strongly consistent reads (within region) if you read the base table; single-item writes are atomic for the base.
  • GSI reads: eventual consistency only. There is no Dynamo feature to issue a strongly consistent query against a GSI — stale results possible shortly after writes.
  • Using TransactWriteItems can atomically update multiple items in base tables, but does not change the GSI’s asynchronous propagation guarantee.
• Cassandra:
  • Materialized views and local secondary indexes: eventual consistency; known anomalies under concurrency.
  • Dedicated index table + LWT for index writes: LWT gives linearizable (serial) consistency for the partition(s) involved in Paxos, but only for those partitions; distributed strong consistency across many partitions requires coordinating multiple LWTs or a separate consensus layer — costly.
  • Reads: you can get strong consistency per partition (QUORUM / LWT guarantees), but cross-partition strong consistency is not free.

Write amplification and performance cost

• DynamoDB GSI:
  • Every write that affects indexed attributes generates extra internal writes to the GSI storage. You pay extra write capacity and storage and may incur higher latency due to internal propagation.
  • Managed retries and backfill can increase write load under hot rebalancing or throughput hotspots.
  • Simpler operationally (AWS manages the complexity) but cost increases with number and size of GSIs.
• Cassandra:
  • Materialized views: additional writes for each base write; since view updates are async, writes may block or be retried, and repair cost increases.
  • Dedicated index table (without LWT): additional synchronous writes to multiple partitions (write amplification equal to number of denormalized copies).
  • Using LWT: significant amplification — Paxos requires multiple round trips to replicas (typically 2–3 phases) so latency and I/O rise sharply; throughput drops and write latency increases.
  • Network/coordination cost grows with number of partitions involved.

Limitations when used across shards/partitions for strongly consistent queries

• DynamoDB GSI:
  • Cannot deliver strongly consistent queries on a GSI. If you need strong consistency across shards, you must:
    • Design indexes so queries are single-partition (hash key) where strong consistency on the base is possible, or
    • Maintain your own index table and update it transactionally (TransactWriteItems across tables) and read the base table with strong consistency; even then GSI reads remain eventually consistent.
  • Cross-shard transactions are supported via TransactWriteItems but are limited (25 items, size limits) and costlier.
• Cassandra:
  • Strong consistency across many partitions is expensive: LWT is per-partition Paxos; to get atomicity across partitions you must orchestrate multi-LWT or application-level two-phase commit, which is slow and complex.
  • Materialized views can diverge under concurrent writes and are not reliable for strong, cross-partition guarantees without heavy repair and operational overhead.
  • Query patterns that require global uniqueness or global serializability across shards are hard to scale; you often need a dedicated coordinator service, external consensus (e.g., ZooKeeper/Etcd), or to restrict the data model so strong guarantees are per-partition.

Concrete recommendations for data engineers

• If you need low operational overhead and can tolerate eventual index staleness, DynamoDB GSI is simple and scalable.
• If you need strict linearizability for an index entry (e.g., global unique constraint), implement a dedicated index with transactional semantics:
  • In DynamoDB: use TransactWriteItems to update base and index tables atomically, but be mindful of Transact limits and that GSIs themselves remain async.
  • In Cassandra: use a dedicated index table plus LWT for the index partition(s) that must be strongly consistent; avoid relying on materialized views for strong guarantees.
• For queries requiring strongly consistent answers across shards at scale, prefer designs that avoid cross-shard serialization (shard by query key, pre-aggregate, or use an external strongly-consistent index service).

Example: enforcing global-unique email

• DynamoDB: best to write a separate table keyed by email and use TransactWriteItems to write both user record and email-key atomically. Avoid relying on a GSI for uniqueness.
• Cassandra: write an index table keyed by email and use LWT (IF NOT EXISTS) on that key to guarantee uniqueness; expect higher write latency and lower throughput.

Summary

• DynamoDB GSI = managed, scalable, eventually consistent index; low ops but no strong reads on GSI and write amplification handled by AWS.
• Cassandra global-index = patterns vary; LWT gives per-partition linearizability but at high cost; materialized views are convenient but can be inconsistent under concurrency; cross-partition strong consistency is expensive and typically avoided by data model choices.

Requirements:

• Validate functional correctness (no lost/duplicated rows), performance (latency, throughput), and safe rollback across workloads (uniform, skewed, hot keys).
• Run in CI for changes to resharding code/infra and as nightly / on-demand suites.
• Integrate with deployment pipelines (GitOps/CD).

High-level architecture:

• Test Orchestrator (Kubernetes Job / Airflow DAG) coordinates scenarios.
• Workload Generator cluster (Kafka producers / custom producer using Spark streaming) emits controlled data patterns.
• Resharding sandbox environment: isolated DB cluster replicas (can be ephemeral cloud instances).
• Metrics & Verification service: collects Prometheus, traces, and row-level checksums.
• Chaos Engine: Litmus/Chaos Mesh to inject failures.
• CI/CD integration: GitHub Actions / GitLab CI triggers orchestration and gates canary rollouts.

Components & steps:

1. Prepare baseline snapshot of source DB and seed expected dataset (N rows).
2. Workload generation:
   • Uniform: random keys across shards.
   • Skewed: Zipf distribution generator, parameterizable skew.
   • Hot-keys: configurable percentage of ops go to small key set.
   • Mix: read/write ratios, TTLs, transaction sizes.
3. Start resharding in sandbox; run workloads concurrently.
4. Automated verification:
   • Row-level checksums: deterministic hashing (PK + payload) computed at source and target; continuous comparator detects missing/duplicated/upsert mismatches.
   • Aggregated metrics: per-shard counts, latency percentiles, throughput, error rates; assert SLAs.
   • End-to-end queries comparing sampled query results.
5. Chaos tests (parameterized):
   • Kill node(s), inject network partition, high GC, I/O latency.
   • Run during different phases: pre-split, during split, post-merge.
6. Canary rollout integration:
   • CI triggers canary plan: reshard small subset (10% traffic) using feature flag/router.
   • Monitor automated verifiers and SLOs for N minutes. If green, widen to 50% then 100%.
   • Use automated rollback if verifiers or chaos-detected anomalies exceed thresholds.
7. Rollback validation:
   • Periodically run rollback rehearsals: perform reshard then automated rollback, verify idempotence and data parity.
   • Validate no metadata drift (partition maps, routing tables).

Automation & CI/CD tips:

• Keep tests declarative (YAML scenarios) so CI can enumerate permutations.
• Store golden checksum snapshots in object storage; use incremental/streaming comparators to avoid full scan.
• Fail-fast and produce machine-readable reports (JSON) for pipeline gates.
• Use feature flags and traffic routers (Envoy/Consul) for canary traffic steering.

Observability and alerts:

• Centralized dashboards (Grafana) with alerting rules tied to verifier anomalies.
• Postmortem artifacts: packet captures, JVM heap dumps, and verifier diffs uploaded on failures.

This framework enables reproducible validation for resharding code paths, automated canary promotion, and safe, testable rollback.

1. Triage & detection

• Scope: identify affected cluster, time window (partition start/end), services using replicas.
• Fast detection techniques:
  • Compare per-partition checksums (Merkle trees / RocksDB SST hashes) between replicas.
  • Use change-data-capture (CDC) / write-ahead-log timestamps and vector-clocks to find divergent keys.
  • Run sampling queries for business-critical namespaces and monitor application error logs for key-related exceptions.

2. Isolate & stop damage

• Stop automated repairs that may write more conflicting state.
• Put read-only mode for impacted datasets (if feasible) or route new writes to a quorum-safe leader.

3. Find divergent keys (practical steps)

• Generate key-level digests per shard and diff them (Merkle trees or per-key CRC).
• Use vector-clock / timestamp comparison: keys with non-comparable vector clocks are conflicts.
• Example quick script pattern (compute hashes and diff):

# compute per-key hash for replica snapshot
for key, value in snapshot.items():
    print(key, hashlib.sha256(value).hexdigest())
# diff outputs across replicas to list divergent keys

4. Reconciliation strategies (choose per data semantics)

• CRDTs: prefer if application can model data as commutative/associative ops (counters, sets). Automates conflict-free merge.
• Last-Writer-Wins (LWW): simple; safe only if clocks are reliable and lost updates acceptable.
• Operational reconciliation scripts: custom resolution where business rules matter (e.g., prefer non-null, sum metrics, merge lists deduped).
• Mixed: tombstones awareness for deletes; use vector-clocks to avoid resurrecting deleted records.
• Automated toolchain: anti-entropy jobs (Merkle-based), Spark jobs to merge snapshots with business logic, or DB-provided repair (Cassandra nodetool repair with incremental, riak_repair-like).

5. Execution plan

• Stage: run reconciliation on a copy; validate with application tests and checksum re-comparison.
• Gradual rollout: reconcile low-risk namespaces first, monitor metrics and application errors.
• Audit: produce a reconciliation log with before/after diffs and provenance (which rule applied).

6. Prevent recurrence

• Improve detection: enable Merkle/anti-entropy, alerting on checksum divergence and elevated conflict rates.
• Strengthen replication: quorum writes/reads, reduce risk of split-brain (use fencing tokens, leader election timeouts).
• Adopt stronger conflict models: migrate critical datasets to CRDTs or single-writer/sharded owner model.
• Clock hygiene: NTP/Chrony, or use logical clocks/vector-clocks.
• Add automated tests (chaos/partition testing) in CI to exercise conflict handling.

7. Customer communication

• Immediate: acknowledge incident, scope (datasets, timeframe), and mitigations in-flight.
• Interim: provide ETA for fixes and recommended workarounds (e.g., re-run affected ETL jobs).
• Final: describe root cause, keys/namespaces impacted, reconciliation approach, validation evidence, and mitigation to prevent recurrence. Offer data rollback or manual support for critical customers and provide SLA/compensation if appropriate.

This approach balances low-risk validation, business-aware reconciliation, automated tooling, and clear stakeholder communication.

Situation: Resharding a 100TB dataset is a large-scale migration problem — you must balance availability, consistency, throughput, and operational risk. Below I compare four common techniques and give recommendations.

1. Copy-based (copy then switch)

• Approach: Copy data from old shards to new shard layout, verify, then atomically switch traffic (e.g., update routing/metadata).
• Reads/Writes during migration: Reads served from old shards; writes need dual-write or write-ahead capture to ensure changes during copy are applied to new shards.
• Downtime: Low if switch is atomic and coordinated; brief window for switch metadata update (seconds-minutes).
• Complexity: Medium-high — implement reliable change capture (CDC), verification, and final cutover.
• Suitability for 100TB: Good if you can stream CDC and copy in parallel; needs substantial bandwidth and storage.

2. Scatter-gather

• Approach: On each request, query both old and new shards (scatter), merge responses (gather), write to both until migration completes.
• Reads/Writes: Reads can be merged; writes are duplicated to both. Ensures correctness but doubles I/O.
• Downtime: Near-zero.
• Complexity: High — client/proxy logic to merge, de-duplicate, and resolve conflicts.
• Suitability for 100TB: Risky at scale due to doubled load and latency; used only for short transitions or low QPS workloads.

3. Redirect-based (proxy forwarding)

• Approach: Proxy forwards requests for moved keys: proxy asks new shard and, if not present, forwards to old shard and optionally backfills.
• Reads/Writes: Reads may incur extra latency for forwarded requests; writes forwarded to new target (or proxied).
• Downtime: Near-zero.
• Complexity: Medium — implement proxy routing rules, timeouts, and backfill mechanics.
• Suitability for 100TB: Good compromise; minimizes immediate bulk copy but increases tail latency and proxy state. Works well with warm-up/backfill.

4. Consistent-hash incremental (rehash by gradually moving key ranges)

• Approach: Incrementally change hash ring; move small subsets at a time and rely on routing layer to target proper node.
• Reads/Writes: Minimal disruption when incremental; write routing must be updated; consistency depends on routing correctness.
• Downtime: None if routing is dynamic and atomic per step.
• Complexity: Medium — requires robust consistent-hash ring management and client/proxy awareness.
• Suitability for 100TB: Recommended if your system already uses consistent hashing; allows gradual migration with controlled bandwidth and validation.

Recommendation for 100TB:

• Prefer incremental approaches (consistent-hash incremental or copy-then-switch with CDC) to avoid huge bursts. If you already have a proxy layer, use redirect-based plus asynchronous bulk copy for backfill. Ensure: robust CDC, monitoring, throttling of re-sharding bandwidth, integrity checks, replay-able cutover, and runbooks for rollback. Plan capacity for dual-write windows and test on smaller datasets first.

Requirements & constraints:

• p95 reads <10ms, high write throughput, user-profile store (hot updates, frequent reads), expect skewed access (some power users).

Shard-key choice:

• Use user_id (stable, uniformly hashed) as primary shard key. If queries often by email or username, maintain a lookup index mapping those to user_id in a small low-latency service or cache.

Consistent hashing vs modulo sharding:

• Start with consistent hashing (virtual nodes) to reduce data movement when nodes added/removed and to handle node churn. Virtual nodes allow balancing without full reshuffle.
• Modulo sharding (user_id % N) is simpler and slightly faster for routing; use it only when cluster size changes are rare and you can afford full re-shard during maintenance.

Replication factor for reads:

• Use RF = 3 across AZs: one primary for writes, two async/semisync replicas for reads. Route reads to nearest replica; use quorum reads for stronger consistency when needed. Serve most reads from read-replicas + edge cache (Redis/CDN) to meet <10ms.

Re-sharding strategies:

• Online re-sharding via consistent-hash ring or use background scatter-gather migration with rate-limited bulk transfer and dual-write period (write to old and new shard) followed by cutover.
• Automate monitoring of shard sizes and CPU/I/O; trigger split when > threshold (e.g., 2x mean).

Handling hot keys:

• Detect hot users via metrics; apply targeted strategies:
  • Move hot key to dedicated shard/instance.
  • Use per-key caching (Redis with TTL) and write-through or change-data-capture to keep cache fresh.
  • Shard within user (sub-sharding) only if profile grows massive (e.g., activity logs).

Cross-shard transactions/queries:

• Avoid multi-shard transactions for reads—denormalize or maintain materialized views for common joins.
• For necessary multi-shard ACID updates, use two-phase commit or application-level compensation with idempotent operations and sagas; prefer eventual consistency.
• Provide a lightweight coordinator service to orchestrate multi-shard operations and record status for retries.

Trade-offs and monitoring:

• Consistent hashing reduces re-shard cost but requires more routing logic.
• Cache aggressively for latency; ensure cache invalidation on writes.
• Monitor p95, replica lag, shard skew, and implement autoscaling and alerting for hot shards.