SQS Sink Node

The sqssink node sends pipeline records as messages to an Amazon SQS queue using the batch send API. Each record becomes one SQS message. Both standard and FIFO queues are supported.

The sink uses the AWS SDK v2 and authenticates via explicit credentials or the default AWS credential chain (environment variables, instance profile, etc.).

Key Features

• Batch sending: up to 10 messages per SQS SendMessageBatch call for efficiency
• FIFO queue support: optionally compute MessageGroupId and MessageDeduplicationId from record fields via path expressions
• JSON encoding: message bodies are serialized as JSON objects (encodingType is required and validated)

Configuration

Field	Type	Required	Default	Description
queueUrl	String	Yes	—	Full SQS queue URL, e.g. https://sqs.us-east-1.amazonaws.com/123456789/my-queue
regionStr	String	Yes	—	AWS region, e.g. us-east-1
encodingType	String	Yes	—	Required and validated against the supported Encoding Types; note that message bodies are always serialized as JSON objects
credentialId	String	No	—	ID of AWS credentials in jobContext.otherProperties. Omit to use default credential chain
messageGroupIdExpression	String	For FIFO queues	—	Path expression evaluated against each record to produce the FIFO MessageGroupId. Required when queueUrl ends with .fifo
deduplicationIdExpression	String	No	—	Path expression evaluated against each record to produce the MessageDeduplicationId
batchSize	Integer	No	10	Messages per SendMessageBatch call (max 10, as per SQS limit)

FIFO Queues

For FIFO queues (queue name ending in .fifo), set messageGroupIdExpression to a path expression that returns a string group identifier. Optionally set deduplicationIdExpression if content-based deduplication is not enabled on the queue.

A deduplication ID represents the identity of a logical event: retries of that event should keep the same ID, while distinct events should receive different IDs unless they are intentionally treated as duplicates. Do not use now(), current timestamps, random values, counters, or other generated-at-send values as a deduplication ID: they can change on retry and do not identify the logical event. Prefer an existing stable event identifier; if none is available, define the event identity before deriving an ID or use content-based deduplication.

config:
  queueUrl: "https://sqs.us-east-1.amazonaws.com/123456789012/alerts.fifo"
  regionStr: "us-east-1"
  encodingType: "JSON_OBJECT"
  messageGroupIdExpression: "$.tenant_id"
  deduplicationIdExpression: "$.event_id"

For standard queues, omit both FIFO fields.

DAG Example

jobContext:
  otherProperties:
    aws-cred:
      username: AKIAIOSFODNN7EXAMPLE
      password: wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY
  metricTags: {}
  dlqConfig:

dag:
  - id: "source"
    commandName: "stdin"
    config:
      encodingType: "JSON_OBJECT"
    outputs:
      - "sink"

  - id: "sink"
    commandName: "sqssink"
    config:
      queueUrl: "https://sqs.us-east-1.amazonaws.com/123456789012/my-queue"
      regionStr: "us-east-1"
      encodingType: "JSON_OBJECT"
      credentialId: "aws-cred"

IAM Permissions

The AWS identity used must have sqs:SendMessage on the target queue:

{
  "Effect": "Allow",
  "Action": [
    "sqs:SendMessage",
    "sqs:GetQueueUrl"
  ],
  "Resource": "arn:aws:sqs:us-east-1:123456789012:my-queue"
}

Related Nodes

• sqssource: Read messages from an SQS queue
• kafkasink: Alternative stream sink for Kafka topics