faucetstream
Get started

Connectors / Databases

Amazon DynamoDB

sink

Amazon DynamoDB sink; batched writes, upsert/delete by key, conditional writes

Install

cargo install faucet-cli --features sink-dynamodb
faucet init --sink dynamodb
# Cargo.toml — use it as a library
cargo add faucet-sink-dynamodb

Starter config

sink:
  type: dynamodb
  config:
    table_name: ""    # REQUIRED — Table name.
    # region: ""    # AWS region.
    # endpoint_url: ""    # Custom endpoint URL (DynamoDB Local, LocalStack, VPC endpoints).
    # credentials: { type: default }    # one of: default, profile, access_key, assume_role, web_identity
      # --- Alternative variants — replace the block above with one of these ---
      # type: profile
      # config:
        # name: ""    # REQUIRED
      #
      # type: access_key
      # config:
        # access_key_id: ""    # REQUIRED
        # secret_access_key: ""    # REQUIRED
        # session_token: ""
      #
      # type: assume_role
      # config:
        # role_arn: ""    # REQUIRED
        # session_name: ""
        # external_id: ""
      #
      # type: web_identity
    # write_mode: null    # Write semantics for a sink.
    # key: []    # Key columns.
    # delete_marker: null    # Optional.
    # rollback: null    # Runtime rollback settings (#706): the run id to journal under, the run-id column, and whether to journal before-images / keep a replaced table.
    # batch_size: 25    # Items per `BatchWriteItem` request (1–25; `0` means 25).
    # concurrency: 4    # Concurrent requests.
    # condition_expression: ""    # Optional `ConditionExpression` applied to every put/delete (for example `attribute_not_exists(pk)` to make re-delivery idempotent).
    # expression_attribute_names: {}    # `ExpressionAttributeNames` for `condition_expression`.
    # expression_attribute_values: {}    # `ExpressionAttributeValues` (plain JSON) for `condition_expression`.
    # on_condition_failure: null    # What a failed `condition_expression` means.
    # retry: { ... }    # Retry/backoff settings exposed on both connector configs.
    # value_format: null    # How records carry attribute values: `plain` JSON (the default) or DynamoDB's `typed` attribute JSON (`{"pk": {"S": "a"}, "tags": {"SS": [..]}}`).

Capabilities

Delivery Effectively-once (keyed upsert)
Write modes append, upsert, delete
Exactly-once writes No
Schema evolution No
Scoped cleanup No
Staged bulk load No
Run rollback No
Compression No

Config fields

FieldTypeDefaultDescription
table_name
required		 string — Table name. The table must exist.
region string — AWS region. None uses the SDK default chain.
endpoint_url string — Custom endpoint URL (DynamoDB Local, LocalStack, VPC endpoints).
credentials object type: default AWS credentials. Defaults to the SDK default provider chain.
write_mode object append Append (default), upsert, or delete.
key array [] Key columns. Required and non-empty for upsert/delete; ignored for append.
delete_marker object — Optional. Upsert only: rows whose field matches one of values are deletes; all others are upserts. The marker field is stripped from upsert rows before writing.
rollback object — Runtime rollback settings (#706): the run id to journal under, the run-id column, and whether to journal before-images / keep a replaced table. Injected per invocation by the CLI when a config has a top-level rollback: block — not meant to be written by hand.
batch_size integer 25 Items per BatchWriteItem request (1–25; 0 means 25). Default 25.
concurrency integer 4 Concurrent requests. Default 4.
condition_expression string — Optional ConditionExpression applied to every put/delete (for example attribute_not_exists(pk) to make re-delivery idempotent). BatchWriteItem cannot carry conditions, so a conditional sink writes item by item.
expression_attribute_names object — ExpressionAttributeNames for condition_expression.
expression_attribute_values object — ExpressionAttributeValues (plain JSON) for condition_expression.
on_condition_failure object skip What a failed condition means. Default skip.
retry object max_retries: 8initial_backoff_ms: 100max_backoff_ms: 10000 Throttle / unprocessed-item retry policy.
value_format object plain How records carry attribute values: plain JSON (default; arrays become lists, base64 strings stay strings) or typed DynamoDB attribute JSON ({"S": "a"}, {"SS": ["x"]}, {"B": "<base64>"}), written exactly as given — pair it with a value_format: typed DynamoDB source to copy sets, binaries and exact numbers losslessly. In typed mode every attribute, keys included, must be a typed value; a delete_marker field may be {"S": "<value>"} or a plain string.
Get started

Your first pipeline runs in five minutes.

Install the CLI, scaffold a config, and move real data — nothing external to stand up.

curl -LsSf https://github.com/faucet-hq/faucet-stream/releases/latest/download/faucet-cli-installer.sh | sh
brew install faucet-hq/faucet-stream/faucet-cli
Read the guide → Star on GitHub