Skip to main content

Amazon SQS

Overview​

Amazon SQS is AWS's managed message queue service. Producers send messages to a queue, consumers poll it, and messages that keep failing move to a dead-letter queue.

The sqs source plugin catalogs the queues of one AWS region as datasets, grouped in a region container, with queue settings as properties, AWS resource tags as DataHub tags, lineage from each queue to its dead-letter queue, and stateful deletion. Its companion, the sns source, catalogs SNS topics and adds topic-to-queue lineage; both build URNs the same way so their lineage joins.

Concept Mapping​

Source ConceptDataHub ConceptNotes
AWS accountPlatform InstanceDefaults to the account id read from a queue ARN
AWS regionContainerSubtype Region
QueueDatasetPlatform sqs, subtype Queue; name <region>.<queue name>
Redrive policyUpstream lineage (patch)Queue to its dead-letter queue
Resource tagTagKey:Value, or Key for an empty value

Module sqs​

Alpha

Important Capabilities​

CapabilityStatusNotes
Asset Containers✅Region containers.
Descriptions❌SQS has no description.
Detect Deleted Entities✅Via stateful ingestion.
Domains✅Supported via the domain config field.
Extract Tags✅From AWS resource tags, added without removing UI tags.
Platform Instance✅Defaults to the AWS account id.
Schema Metadata❌SQS has no schema.
Table-Level Lineage✅Queue to dead-letter queue.
Test Connection✅Enabled by default.

Overview​

The sqs module reads the SQS control-plane APIs for one AWS region per recipe. It never receives or deletes messages, so running it has no effect on consumers.

SQS has no schema, so queues carry no schema fields. Queue settings (FIFO, visibility timeout, retention, encryption, dead-letter target, maximum receive count) are stored as custom properties, and the creation and last-modified times on the dataset properties.

SNS topics are ingested by the separate sns source. Run both with the same env, platform_instance and account_platform_instance_map so that topic-to-queue lineage written by sns lands on the queues this source emits.

Prerequisites​

AWS IAM Permissions​

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "SqsRead",
      "Effect": "Allow",
      "Action": [
        "sqs:ListQueues",
        "sqs:GetQueueAttributes",
        "sqs:ListQueueTags"
      ],
      "Resource": "*"
    },
    {
      "Sid": "TagsRead",
      "Effect": "Allow",
      "Action": ["tag:GetResources"],
      "Resource": "*"
    }
  ]
}

With extract_tags on, the connector reads every queue's tags with one paginated tag:GetResources call. If that call is denied it warns and falls back to sqs:ListQueueTags per queue, so either permission works.

Authentication​

Credentials come from aws_config or the standard boto3 chain: environment variables, a shared credentials profile, an IAM role on the host, or SSO. aws_config.aws_role assumes a role, including one in another account.

Install the Plugin​

pip install 'acryl-datahub[sqs]'

Starter Recipe​

Check out the following recipe to get started with ingestion! See below for full configuration options.

For general pointers on writing and running a recipe, see our main recipe guide.

source:
  type: sqs
  config:
    aws_config:
      aws_region: "us-east-1"
      # aws_role: "arn:aws:iam::123456789012:role/datahub-read"
    # platform_instance defaults to the AWS account id. Use the same env and
    # platform_instance in the sns recipe so topic-to-queue lineage joins.
    env: "PROD"

    queue_pattern:
      deny:
        - ".*-tmp$"

    extract_tags: true

    # domain:
    #   "urn:li:domain:payments":
    #     allow:
    #       - "orders.*"

    # Platform instances used by other accounts' recipes, for cross-account lineage.
    # account_platform_instance_map:
    #   "210987654321": "shared-services"

    stateful_ingestion:
      enabled: true

Config Details​

Note that a . is used to denote nested fields in the YAML recipe.

FieldDescription
account_platform_instance_map
map(str,string)
extract_tags
boolean
Extract AWS resource tags as DataHub tags. SNS allows 10 ListTagsForResource calls per second, so turn this off on the sns source for accounts with tens of thousands of topics.
Default: True
platform_instance
One of string, null
The instance of the platform that all assets produced by this recipe belong to. This should be unique within the platform. See https://docs.datahub.com/docs/platform-instances/ for more details.
Default: None
remove_stale_lineage
boolean
Remove lineage edges this source wrote on an earlier run that no longer exist in AWS, such as a removed redrive policy or subscription. Only edges from resources this run read in full are removed. Needs a DataHub graph connection (the datahub-rest sink or datahub_api); costs one DataHub lookup per queue or topic.
Default: True
env
string
The environment that all assets produced by this connector belong to
Default: PROD
aws_config
AwsConnectionConfig
Common AWS credentials config.

Currently used by:
- Glue source
- SageMaker source
- dbt source
aws_config.aws_access_key_id
One of string, null
AWS access key ID. Can be auto-detected, see the AWS boto3 docs for details.
Default: None
aws_config.aws_advanced_config
object
Advanced AWS configuration options. These are passed directly to botocore.config.Config.
aws_config.aws_endpoint_url
One of string, null
The AWS service endpoint. This is normally constructed automatically, but can be overridden here.
Default: None
aws_config.aws_profile
One of string, null
The named profile to use from AWS credentials. Falls back to default profile if not specified and no access keys provided. Profiles are configured in ~/.aws/credentials or ~/.aws/config.
Default: None
aws_config.aws_proxy
One of string, null
A set of proxy configs to use with AWS. See the botocore.config docs for details.
Default: None
aws_config.aws_region
One of string, null
AWS region code.
Default: None
aws_config.aws_retry_mode
Enum
One of: "legacy", "standard", "adaptive"
Default: standard
aws_config.aws_retry_num
integer
Number of times to retry failed AWS requests. See the botocore.retry docs for details.
Default: 5
aws_config.aws_secret_access_key
One of string(password), null
AWS secret access key. Can be auto-detected, see the AWS boto3 docs for details.
Default: None
aws_config.aws_session_token
One of string(password), null
AWS session token. Can be auto-detected, see the AWS boto3 docs for details.
Default: None
aws_config.read_timeout
number
The timeout for reading from the connection (in seconds).
Default: 60
aws_config.aws_role
One of string, array, null
AWS roles to assume. If using the string format, the role ARN can be specified directly. If using the object format, the role can be specified in the RoleArn field and additional available arguments are the same as boto3's STS.Client.assume_role.
Default: None
aws_config.aws_role.union
One of string, AwsAssumeRoleConfig
aws_config.aws_role.union.RoleArn ❓
string
ARN of the role to assume.
aws_config.aws_role.union.ExternalId
One of string, null
External ID to use when assuming the role.
Default: None
domain
map(str,AllowDenyPattern)
A class to store allow deny regexes.

Patterns are matched against the start of the string only, not the entire
string - a pattern does not need to match to the end to be considered a match.
For example, the pattern "prod" matches "prod", "prod_east", and "production".
To require an exact match, anchor your pattern explicitly, e.g. "^prod$".
domain.key.allow
array
List of regex patterns to include in ingestion. Patterns match from the start of the string only, not the entire string - anchor with '^...$' for an exact match, e.g. '^prod$'.
Default: ['.*']
domain.key.allow.string
string
domain.key.ignoreCase
One of boolean, null
Whether to ignore case sensitivity during pattern matching.
Default: True
domain.key.deny
array
List of regex patterns to exclude from ingestion. Patterns match from the start of the string only, not the entire string - anchor with '^...$' for an exact match, e.g. '^prod$'.
Default: []
domain.key.deny.string
string
queue_pattern
AllowDenyPattern
A class to store allow deny regexes.

Patterns are matched against the start of the string only, not the entire
string - a pattern does not need to match to the end to be considered a match.
For example, the pattern "prod" matches "prod", "prod_east", and "production".
To require an exact match, anchor your pattern explicitly, e.g. "^prod$".
queue_pattern.ignoreCase
One of boolean, null
Whether to ignore case sensitivity during pattern matching.
Default: True
stateful_ingestion
One of StatefulStaleMetadataRemovalConfig, null
Stateful ingestion configuration for stale entity removal.
Default: None
stateful_ingestion.enabled
boolean
Whether or not to enable stateful ingest. Default: True if a pipeline_name is set and either a datahub-rest sink or datahub_api is specified, otherwise False
Default: False
stateful_ingestion.fail_safe_threshold
number
Prevents large amount of soft deletes & the state from committing from accidental changes to the source configuration if the relative change percent in entities compared to the previous state is above the 'fail_safe_threshold'.
Default: 75.0
stateful_ingestion.remove_stale_metadata
boolean
Soft-deletes the entities present in the last successful run but missing in the current run with stateful_ingestion enabled.
Default: True

Capabilities​

Platform instance and URNs​

When platform_instance is not set, the connector reads the account id from a queue ARN and uses it. Dataset names include the region, so several regions of one account share the platform instance without colliding. The sns source names topics the same way.

Dead-letter queue lineage​

Each queue with a redrive policy becomes an upstream of its dead-letter queue, with a relationship: dead_letter property on the edge. Edges are written as patches, so lineage that the sns source or another account's recipe adds to the same queue is kept. The edge is only added when the dead-letter queue was ingested in the same run; one excluded by queue_pattern gets none. Skipped edges are listed in the report under lineage_edges_skipped.

When a redrive policy is removed or pointed at another queue, the next run removes the old dead_letter edge (remove_stale_lineage, on by default). This reads the queue's current lineage from DataHub, so it needs a graph connection (the datahub-rest sink or datahub_api); without one, old edges stay. Only dead_letter edges from queues read in full this run are removed, so a queue whose attributes could not be read keeps its edges. Edges with other relationships, such as the sns source's subscription edges, and edges from queues that other recipes ingest are never touched.

Tags​

AWS resource tags become DataHub tags named Key:Value (or Key when the value is empty), the same format as the kinesis source. They are added to the queue rather than replacing its tags, so tags added in the DataHub UI stay. The flip side is that a tag removed in AWS stays in DataHub until removed there. To turn an owner tag into ownership, add the extract_ownership_from_tags transformer to the recipe.

Domains​

domain maps a domain urn or id to regex patterns on queue names, as in the glue source. A queue matching a pattern is assigned that domain.

Limitations​

  • One region per recipe. Run one recipe per region for multi-region accounts.
  • No schema metadata: SQS stores opaque message bodies.
  • Queue depth (ApproximateNumberOfMessages and similar) is not ingested; it changes every few seconds.

Troubleshooting​

  • "Listing denied" failure: the role lacks sqs:ListQueues. Add the statement above.
  • "Listing aborted mid-pagination" failure: AWS throttled or failed partway through the listing. Re-run before relying on stale-entity removal, since queues past that point were not seen.
  • "Could not read a resource ARN to find the AWS account id" error: the first sqs:ListQueues call failed, so the run stopped rather than write queues under URNs without the account id. Re-run, or set platform_instance explicitly.

Code Coordinates​

  • Class Name: datahub.ingestion.source.aws_messaging.sqs_source.SqsSource
  • Browse on GitHub
Questions?

If you've got any questions on configuring ingestion for Amazon SQS, feel free to ping us on our Slack.

💡 Contributing to this documentation

This page is auto-generated from the underlying source code. To make changes, please edit the relevant source files in the metadata-ingestion directory.

Tip: For quick typo fixes or documentation updates, you can click the ✏️ Edit icon directly in the GitHub UI to open a Pull Request. For larger changes and PR naming conventions, please refer to our Contributing Guide.