Skip to main content

S3 Realtime Source Node

The s3rtsource node streams S3 objects into the pipeline as they arrive. Instead of listing a bucket, it long-polls an Amazon SQS queue that receives S3 event notifications, downloads each newly created object, deserializes its contents, and emits the resulting records. SQS messages are deleted only after their objects are successfully processed.

The source uses the AWS SDK v2 and authenticates via explicit credentials or the default AWS credential chain (environment variables, instance profile, etc.). The same credential is used for both SQS and S3 by default, with optional overrides for cross-account buckets. Only ObjectCreated:* events are processed; s3:TestEvent and other event types are acknowledged and skipped.

Key Features​

  • Event-driven ingestion: reads objects as S3 emits ObjectCreated notifications — no bucket scanning or polling of S3 itself
  • SQS long polling: reduces empty receive calls and latency using configurable wait time (up to 20 seconds)
  • At-most-once download per object: objects are downloaded one at a time, so at most a single object is held in memory
  • Compression auto-detection: gzip is auto-detected from the object's magic bytes; can be forced with compressionType: GZIP
  • Object size guard: objects larger than maxObjectSizeBytes are dead-lettered without being downloaded
  • Retry cap and dead-lettering: malformed notifications, unparseable objects, and messages exceeding maxRetries are dead-lettered rather than retried forever
  • Optional S3 metadata: enrich each record with its source __s3_bucket and __s3_key
  • Infinite streaming: never exhausts — runs until the pipeline is terminated

Configuration​

FieldTypeRequiredDefaultDescription
queueUrlStringYes—Full URL of the SQS queue receiving the S3 event notifications, e.g. https://sqs.us-east-1.amazonaws.com/123456789012/my-queue
regionStrStringYes—AWS region of the SQS queue, e.g. us-east-1
encodingTypeStringYes—Encoding of the referenced S3 objects' content. See Encoding Types
compressionTypeStringNonullCompression of the referenced objects. When omitted, gzip is auto-detected from the object's magic bytes (both plain and gzipped objects work); set GZIP to force decompression
credentialIdStringNo—ID of AWS credentials in jobContext.otherProperties, used for both SQS and S3. Omit to use the default AWS credential chain
s3CredentialIdStringNocredentialIdOptional S3 credential override for cross-account buckets
s3RegionStrStringNoregionStrOptional S3 region override (when the bucket lives in a different region than the queue)
s3EndpointOverrideStringNonullOptional custom S3-compatible endpoint URL (e.g. MinIO, LocalStack)
maxNumberOfMessagesintNo10Messages to retrieve per SQS poll (1–10, as per SQS limit)
waitTimeSecondsintNo20SQS long polling wait time in seconds (0–20)
visibilityTimeoutSecondsintNo300How long a received message is hidden from other consumers. A message is deleted only after its object is fully downloaded, decompressed, and deserialized; if that takes longer than this window the message becomes visible again and is reprocessed, causing duplicate emits. Size this above the expected worst-case processing time
maxObjectSizeByteslongNo268435456 (256 MiB)Objects whose size exceeds this are dead-lettered (with the notification and reason) and acknowledged; their content is not downloaded
maxRetriesintNo5In-app retry cap: messages whose SQS ApproximateReceiveCount exceeds this are dead-lettered
addS3MetadatabooleanNofalseWhen true, enrich each emitted record with __s3_bucket and __s3_key

How It Works​

This source does not scan S3. It relies on S3 event notifications delivered through an SQS queue:

  1. An object is created in the S3 bucket.
  2. S3 sends an ObjectCreated event notification to the configured SQS queue (either directly, or via an SNS topic that fans out to SQS — both envelope shapes are parsed).
  3. The source long-polls the queue, parses each notification into its referenced bucket/key(s), and downloads the object(s).
  4. Each object is decompressed (if needed), deserialized using encodingType, and emitted downstream.
  5. On success the SQS message is deleted. On terminal failure the notification is dead-lettered and acknowledged.

Failure handling​

  • Malformed notification envelope: dead-lettered and deleted (retrying can never succeed).
  • Deleted object (NoSuchKey): skipped with no dead-letter — e.g. when a lifecycle policy expired the object before it was read.
  • Oversized object (> maxObjectSizeBytes): dead-lettered (with reason) and acknowledged; body is never downloaded.
  • Unparseable object content: dead-lettered once and acknowledged on the first attempt — no retry storm.
  • Exceeded maxRetries: message dead-lettered and deleted based on its ApproximateReceiveCount.

Dead-lettering requires a dlqConfig on the job context; without one, failed messages are logged and dropped.

AWS Setup​

To use this source you must configure S3 to publish ObjectCreated events to an SQS queue.

  1. Create an SQS queue and note its URL (see Getting the Queue URL).
  2. Grant S3 permission to send to the queue with an SQS queue policy allowing the s3.amazonaws.com principal sqs:SendMessage on the queue ARN.
  3. Enable bucket notifications on the source bucket for the s3:ObjectCreated:* event type, targeting the queue.

Optionally, an SNS topic can sit between S3 and SQS (S3 → SNS → SQS); the source unwraps the SNS envelope automatically.

Getting the Queue URL​

The full queue URL is required — not just the queue name. Retrieve it with:

aws sqs get-queue-url --queue-name my-queue

Output:

{
    "QueueUrl": "https://sqs.us-east-1.amazonaws.com/123456789012/my-queue"
}

DAG Example​

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

dag:
  - id: "source"
    commandName: "s3rtsource"
    config:
      queueUrl: "https://sqs.us-east-1.amazonaws.com/123456789012/my-queue"
      regionStr: "us-east-1"
      encodingType: "JSON_OBJECT_LINE"
      credentialId: "aws-cred"
      maxNumberOfMessages: 10
      waitTimeSeconds: 20
      visibilityTimeoutSeconds: 300
      addS3Metadata: true
    outputs:
      - "sink"

  - id: "sink"
    commandName: "stdout"
    config:
      encodingType: "JSON_OBJECT"

IAM Permissions​

The AWS identity used must have SQS permissions on the queue and S3 read permission on the source bucket:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "sqs:ReceiveMessage",
        "sqs:DeleteMessage",
        "sqs:GetQueueAttributes"
      ],
      "Resource": "arn:aws:sqs:us-east-1:123456789012:my-queue"
    },
    {
      "Effect": "Allow",
      "Action": "s3:GetObject",
      "Resource": "arn:aws:s3:::my-bucket/*"
    }
  ]
}

When s3CredentialId is set for a cross-account bucket, the s3:GetObject permission applies to that account's credential.

  • s3sink: Write records to Amazon S3 (or S3-compatible storage)
  • sqssource: Read raw messages directly from an SQS queue
  • kafkasource: Alternative stream source for Kafka topics