dynamodb

Author: Meroxa, Inc.

1

Latest release

• conduit-connector-dynamodb_0.4.0_Darwin_arm64.tar.gz
• conduit-connector-dynamodb_0.4.0_Darwin_x86_64.tar.gz
• conduit-connector-dynamodb_0.4.0_Linux_arm64.tar.gz
• conduit-connector-dynamodb_0.4.0_Linux_i386.tar.gz
• conduit-connector-dynamodb_0.4.0_Linux_x86_64.tar.gz
• conduit-connector-dynamodb_0.4.0_Windows_arm64.tar.gz
• conduit-connector-dynamodb_0.4.0_Windows_i386.tar.gz
• conduit-connector-dynamodb_0.4.0_Windows_x86_64.tar.gz

Description

Source

A source connector that pulls data from a DynamoDB table to downstream resources via Conduit.

The connector starts with a snapshot of the data currently existent in the table, sends these records to the destination, then starts the CDC (Change Data Capture) mode which will listen to events happening on the table in real-time, and sends these events' records to the destination (these events include: updates, deletes, and inserts). You can opt out from taking the snapshot by setting the parameter skipSnapshot to true, meaning that only the CDC events will be captured.

The source connector uses DynamoDB Streams to get CDC events, so you need to enable the stream before running the connector. Check out the documentation for how to enable a DynamoDB Stream.

Destination

A destination connector that takes a conduit record and stores it into a DynamoDB table.

The Destination is designed to handle different kinds of operations, based on the Operation field in the record received, the destination will either insert, update or delete the record in the target table. Snapshot records are always inserted.

Note: Make sure to create the destination table before running the pipeline, the table needs to have the same schema as the records that will be sent into it, either from the source, or from a processor.

Source Parameters

version: 2.2
pipelines:
  - id: example
    status: running
    connectors:
      - id: example-source
        type: source
        plugin: "dynamodb"
        name: example-source
        settings:
          # AWS region.
          # Type: string
          aws.region: ""
          # Table is the DynamoDB table name to pull data from, or push data
          # into.
          # Type: string
          table: ""
          # AWS access key id. Optional - if not provided, the connector will
          # use the default credential chain (environment variables, shared
          # credentials file, or IAM role). For production environments, it's
          # recommended to use the default credential chain with IAM roles
          # rather than static credentials.
          # Type: string
          aws.accessKeyId: ""
          # AWS secret access key. Optional - if not provided, the connector
          # will use the default credential chain (environment variables, shared
          # credentials file, or IAM role). For production environments, it's
          # recommended to use the default credential chain with IAM roles
          # rather than static credentials.
          # Type: string
          aws.secretAccessKey: ""
          # AWS temporary session token. Optional - if not provided, the
          # connector will use the default credential chain. Note that to keep
          # the connector running long-term, you should use the default
          # credential chain rather than temporary session tokens which will
          # expire. For production environments, it's recommended to use IAM
          # roles (IRSA, EC2 instance profile, or ECS task role).
          # Type: string
          aws.sessionToken: ""
          # The URL for AWS (useful when testing the connector with localstack).
          # Type: string
          aws.url: ""
          # Discovery polling period for the CDC mode of how often to check for
          # new shards in the DynamoDB Stream, formatted as a time.Duration
          # string.
          # Type: duration
          discoveryPollingPeriod: "10s"
          # Records polling period for the CDC mode of how often to get new
          # records from a shard, formatted as a time.Duration string.
          # Type: duration
          recordsPollingPeriod: "1s"
          # SkipSnapshot determines weather to skip the snapshot or not.
          # Type: bool
          skipSnapshot: "false"
          # Maximum delay before an incomplete batch is read from the source.
          # Type: duration
          sdk.batch.delay: "0"
          # Maximum size of batch before it gets read from the source.
          # Type: int
          sdk.batch.size: "0"
          # Specifies whether to use a schema context name. If set to false, no
          # schema context name will be used, and schemas will be saved with the
          # subject name specified in the connector (not safe because of name
          # conflicts).
          # Type: bool
          sdk.schema.context.enabled: "true"
          # Schema context name to be used. Used as a prefix for all schema
          # subject names. If empty, defaults to the connector ID.
          # Type: string
          sdk.schema.context.name: ""
          # Whether to extract and encode the record key with a schema.
          # Type: bool
          sdk.schema.extract.key.enabled: "true"
          # The subject of the key schema. If the record metadata contains the
          # field "opencdc.collection" it is prepended to the subject name and
          # separated with a dot.
          # Type: string
          sdk.schema.extract.key.subject: "key"
          # Whether to extract and encode the record payload with a schema.
          # Type: bool
          sdk.schema.extract.payload.enabled: "true"
          # The subject of the payload schema. If the record metadata contains
          # the field "opencdc.collection" it is prepended to the subject name
          # and separated with a dot.
          # Type: string
          sdk.schema.extract.payload.subject: "payload"
          # The type of the payload schema.
          # Type: string
          sdk.schema.extract.type: "avro"

Destination Parameters

version: 2.2
pipelines:
  - id: example
    status: running
    connectors:
      - id: example-destination
        type: destination
        plugin: "dynamodb"
        name: example-destination
        settings:
          # AWS region.
          # Type: string
          aws.region: ""
          # Table is the DynamoDB table name to pull data from, or push data
          # into.
          # Type: string
          table: ""
          # AWS access key id. Optional - if not provided, the connector will
          # use the default credential chain (environment variables, shared
          # credentials file, or IAM role). For production environments, it's
          # recommended to use the default credential chain with IAM roles
          # rather than static credentials.
          # Type: string
          aws.accessKeyId: ""
          # AWS secret access key. Optional - if not provided, the connector
          # will use the default credential chain (environment variables, shared
          # credentials file, or IAM role). For production environments, it's
          # recommended to use the default credential chain with IAM roles
          # rather than static credentials.
          # Type: string
          aws.secretAccessKey: ""
          # AWS temporary session token. Optional - if not provided, the
          # connector will use the default credential chain. Note that to keep
          # the connector running long-term, you should use the default
          # credential chain rather than temporary session tokens which will
          # expire. For production environments, it's recommended to use IAM
          # roles (IRSA, EC2 instance profile, or ECS task role).
          # Type: string
          aws.sessionToken: ""
          # The URL for AWS (useful when testing the connector with localstack).
          # Type: string
          aws.url: ""
          # Maximum delay before an incomplete batch is written to the
          # destination.
          # Type: duration
          sdk.batch.delay: "0"
          # Maximum size of batch before it gets written to the destination.
          # Type: int
          sdk.batch.size: "0"
          # Allow bursts of at most X records (0 or less means that bursts are
          # not limited). Only takes effect if a rate limit per second is set.
          # Note that if `sdk.batch.size` is bigger than `sdk.rate.burst`, the
          # effective batch size will be equal to `sdk.rate.burst`.
          # Type: int
          sdk.rate.burst: "0"
          # Maximum number of records written per second (0 means no rate
          # limit).
          # Type: float
          sdk.rate.perSecond: "0"
          # The format of the output record. See the Conduit documentation for a
          # full list of supported formats
          # (https://conduit.io/docs/using/connectors/configuration-parameters/output-format).
          # Type: string
          sdk.record.format: "opencdc/json"
          # Options to configure the chosen output record format. Options are
          # normally key=value pairs separated with comma (e.g.
          # opt1=val2,opt2=val2), except for the `template` record format, where
          # options are a Go template.
          # Type: string
          sdk.record.format.options: ""
          # Whether to extract and decode the record key with a schema.
          # Type: bool
          sdk.schema.extract.key.enabled: "true"
          # Whether to extract and decode the record payload with a schema.
          # Type: bool
          sdk.schema.extract.payload.enabled: "true"