faucet-source-dynamodb

Amazon DynamoDB source for faucet-stream: parallel scan, query by key condition, and DynamoDB Streams CDC.

source:
  type: dynamodb
  config:
    table_name: orders
    region: us-east-1
    mode: scan          # scan | query | streams
    segments: 8         # parallel scan

Configuration

Items are converted to plain JSON losslessly: numbers that are not exactly representable (beyond i64/u64 or an exact f64) stay decimal strings, sets become arrays, binary becomes base64.

With value_format: typed every attribute instead arrives in DynamoDB's attribute JSON — {"pk": {"S": "a"}, "tags": {"SS": ["x"]}, "blob": {"B": "AQ=="}, "price": {"N": "12345678901234567890.5"}} — in scan/query items and in the Streams envelope's key / before / after images. Paired with a value_format: typed DynamoDB sink this copies a table exactly: sets stay sets, binaries stay B, numbers keep every digit. Transforms and other sinks see the typed objects, so use it for DynamoDB-to-DynamoDB copies (a cdc_unwrap stage works, but the flattened row still carries typed values).

Scan and query

Each segment pages with LastEvaluatedKey. Intermediate pages carry a bookmark with every segment's cursor (dynamodb:<table> state key), so a crashed run resumes each segment where it stopped; the final page resets the cursors, so the next run reads the table again (a scan is a snapshot). Throttling (ProvisionedThroughputExceededException and friends) backs off and retries; consumed capacity is logged per segment.

Cluster sharding

mode: scan is shardable: enumerate_shards returns one shard per segment (segments when set above 1, otherwise the coordinator's target), and each worker scans only its segment.

Streams CDC

mode: streams reads the table's stream (enable it with NEW_AND_OLD_IMAGES). Each change becomes a cdc_unwrap-compatible envelope:

{ "op": "c|u|d", "before": {…} | null, "after": {…} | null, "key": {…},
  "document_key": {…}, "table": "orders", "event_id": "…", "event_name": "INSERT",
  "sequence_number": "…", "shard_id": "…", "ts_ms": 1716700000000,
  "size_bytes": 42, "stream_view_type": "NEW_AND_OLD_IMAGES", "user_identity": null }

TTL expirations arrive as op: "d" with user_identity.principal_id = "dynamodb.amazonaws.com". With KEYS_ONLY / NEW_IMAGE views before is null, and cdc_unwrap falls back to document_key for deletes.

• Ordering — a child shard is read only after its parent is drained; when a shard closes the stream is re-described to pick up its children.
• Scheduling — a pool of shard_concurrency workers rotates over every readable shard: a worker reads a bounded slice of a shard (up to 8 batches, or until it is caught up or throttled), hands it back, and takes the next one. A caught-up shard is polled again after poll_interval_ms, a throttled one after a backoff, so more open shards than workers never stalls any of them.
• Bookmarks — every page carries {stream_arn, shards: {id: seq}, finished} (state key dynamodb-streams:<table>). A shard opened but not yet read is recorded with an empty sequence and resumes from its trim horizon, so nothing between runs is skipped; verified_at records when such a shard was last known to hold nothing unread. A child of a shard that was read always starts at its trim horizon. A latest shard whose iterator expires before its first record is re-read from its trim horizon, keeping records created since about when it was first opened, instead of jumping to a new latest.
• Idle termination — idle_termination_secs counts only time spent waiting for records; time the sink spends writing a page does not end the run.
• Gaps — Streams keep 24 hours of changes. Resuming fails with the gap named when the stream was replaced, a bookmarked shard expired, a shard's parent expired unread, a bookmarked position was trimmed, or a shard bookmarked with no position was last known empty more than 23 hours ago (its trim horizon never reports trimming, so records written and trimmed since would be lost silently). Run a streams pipeline more often than every 23 hours (or continuously): a daily schedule with an idle shard reaches that limit on most runs. A bookmark written before verified_at existed cannot be checked on its first resume — the source logs a warning naming those shards, and the check applies from that run on. With on_gap: resnapshot the source instead scans the whole table (emitted as op: "r" envelopes, no bookmark until the scan completes) and then replays the retained stream from the trim horizon. That run is a complete read (Source::read_completeness): once it ends, the destination rows of items the scan did not see — items deleted while their stream records were trimmed — are deleted through the sink's keyed cleanup, and only then is the new position saved (a crash or failed delete re-reads the table on the next run). An ordinary streams run never deletes. The sink must support cleanup with write_mode: upsert and a key (postgres, mysql, mssql, sqlite, mongodb, elasticsearch, bigquery, spanner); faucet validate refuses on_gap: resnapshot otherwise, unless resnapshot_deletes: false (re-read the table, delete nothing, keep the rows of items deleted during the gap). Up to resnapshot_max_keys (default 1,000,000) distinct keys are tracked; a re-read that writes more fails once with nothing deleted and saves its position, so the next run resumes streaming — see complete reads.
• faucet mirror — capture_resume_position returns every current shard at its trim horizon, so a CDC run after the snapshot replays the retained window and converges.
• Delivery — at-least-once (supports_exactly_once is false, like the Kinesis source). Pair with an upsert sink keyed on the table key.

Source lag

In mode: streams, Source::lag (#733) reports seconds: now minus the ApproximateCreationDateTime of the newest record read from each open shard, the largest across shards. Before this run has read anything — faucet status --probe, faucet doctor, the start of a run — it peeks each open shard from the bookmark instead: GetShardIterator at the bookmarked sequence (or where a run would start for a shard with no entry) and one GetRecords with Limit: 1; the first record's ApproximateCreationDateTime is the age of the oldest unconsumed change. Stream reads are non-destructive, so the probe consumes nothing. DynamoDB Streams exposes no head position, so a shard with nothing to read counts as caught up (zero) rather than aging while quiet. The pipeline polls it on the first page, at most every 15 s after, and when the run ends, exporting faucet_source_lag_seconds{pipeline,row,connector}; sla.max_lag_seconds turns it into an SLA. A failed probe is logged once and reported as no lag — it never fails a run. scan and query modes report no lag.

Discovery and preflight

discover() lists tables (ListTables + DescribeTable): one descriptor per table with its key attributes as the schema, ItemCount as the row estimate and {table_name} as the config patch. check() runs DescribeTable (and DescribeStream in streams mode) — nothing is read.

Capabilities

Declares dataset discovery for every mode; sharding only under mode: scan (parallel segments), lag reporting and table routing only under mode: streams, and complete reads (complete_reads) only under mode: streams with on_gap: resnapshot. faucet list --json prints the declaration.

Timeouts, cancellation and retries

Every DynamoDB and DynamoDB Streams request runs through the pipeline's CallContext (#880): each attempt is bounded by the default 300 s call timeout and any caller deadline, the run's cancel token stops an in-flight request, a backoff and the shard scheduler's poll wait, and each attempt is metered as a round trip (scan / query, describe_table, list_tables, describe_stream, get_shard_iterator, get_records). The throttle-aware retry: policy is unchanged; its backoffs wait through the same context.

License

MIT OR Apache-2.0