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
Lambda function wired to a queueDataFlow + DataJobPlatform lambda; one invoke job per function
EventBridge rule or Pipe touching a queueDataFlow + DataJobPlatforms eventbridge and eventbridge-pipes; event buses are Event Bus datasets
Resource tagTagKey:Value, or Key for an empty value

Module sqs​

Alpha

Important Capabilities​

CapabilityStatusNotes
Asset Containers✅Region containers.
Column-level Lineage✅Queue to dead-letter queue, when both have schemas.
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✅Opt-in: AWS Glue Schema Registry schemas listed in glue_schema_registry.queue_schema_map, or found by use_naming_convention.
Table-Level Lineage✅Dead-letter queues, Lambda triggers and destinations, EventBridge rules and Pipes.
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 of its own; queue schemas can be attached from AWS Glue Schema Registry (opt-in). 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": "GlueSchemaRegistryRead",
      "Effect": "Allow",
      "Action": ["glue:GetSchemaVersion"],
      "Resource": [
        "arn:aws:glue:<region>:<account-id>:registry/<registry-name>",
        "arn:aws:glue:<region>:<account-id>:schema/<registry-name>/*"
      ]
    },
    {
      "Sid": "LambdaRead",
      "Effect": "Allow",
      "Action": ["lambda:ListEventSourceMappings"],
      "Resource": "*"
    },
    {
      "Sid": "LambdaAsyncDestinationsRead",
      "Effect": "Allow",
      "Action": [
        "lambda:ListFunctions",
        "lambda:ListFunctionEventInvokeConfigs"
      ],
      "Resource": "*"
    },
    {
      "Sid": "EventBridgeList",
      "Effect": "Allow",
      "Action": [
        "events:ListEventBuses",
        "events:ListRules",
        "pipes:ListPipes"
      ],
      "Resource": "*"
    },
    {
      "Sid": "EventBridgeRead",
      "Effect": "Allow",
      "Action": ["events:ListTargetsByRule"],
      "Resource": "arn:aws:events:<region>:<account-id>:rule/*"
    },
    {
      "Sid": "PipesRead",
      "Effect": "Allow",
      "Action": ["pipes:DescribePipe"],
      "Resource": "arn:aws:pipes:<region>:<account-id>:pipe/*"
    },
    {
      "Sid": "S3ListBuckets",
      "Effect": "Allow",
      "Action": ["s3:ListAllMyBuckets"],
      "Resource": "*"
    },
    {
      "Sid": "S3NotificationsRead",
      "Effect": "Allow",
      "Action": ["s3:GetBucketNotification"],
      "Resource": "arn:aws:s3:::*"
    },
    {
      "Sid": "CloudTrailLakeQuery",
      "Effect": "Allow",
      "Action": [
        "cloudtrail:StartQuery",
        "cloudtrail:GetQueryResults",
        "cloudtrail:CancelQuery"
      ],
      "Resource": "arn:aws:cloudtrail:<region>:<account-id>:eventdatastore/<event-data-store-id>"
    },
    {
      "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.

glue:GetSchemaVersion is only needed with glue_schema_registry.enabled.

LambdaRead is used with lineage.lambda_functions, and LambdaAsyncDestinationsRead only with lineage.lambda_async_destinations too; leave it out otherwise. EventBridgeList is used with lineage.eventbridge_rules or lineage.eventbridge_pipes, EventBridgeRead with lineage.eventbridge_rules, and PipesRead with lineage.eventbridge_pipes. The ListEventBuses, ListRules and ListPipes actions don't support resource-level scoping, so they need *. pipes:DescribePipe is only called for a pipe that could get column lineage. lineage.lambda_async_destinations is off by default because it makes one lambda:ListFunctionEventInvokeConfigs call per function in the region, from both sources; it stops at the first AccessDenied, and so does the events:ListTargetsByRule loop.

S3ListBuckets and S3NotificationsRead are used only with lineage.s3_notifications, which is off by default because it makes one s3:GetBucketNotification call per bucket in the recipe's region. s3:ListAllMyBuckets only returns buckets the calling account owns, so buckets in other accounts are not read. To read only some buckets, list their ARNs (arn:aws:s3:::my-bucket) in S3NotificationsRead; a bucket whose notifications are denied is skipped with a warning and keeps its earlier edges.

CloudTrailLakeQuery is used only with cloudtrail_producers.enabled; scope it to the store in cloudtrail_producers.event_data_store. cloudtrail:CancelQuery stops a query the run gave up on, which would otherwise keep scanning. Matching a caller to a Lambda function also needs lambda:ListFunctions, so with cloudtrail_producers.enabled grant that action even when LambdaAsyncDestinationsRead is otherwise left out.

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"

    glue_schema_registry:
      enabled: false
      registry_name: "default-registry"
      queue_schema_map: {}
      #   orders: "order-v1"

    lineage:
      lambda_functions: true
      # One lambda:ListFunctionEventInvokeConfigs call per function in the region.
      lambda_async_destinations: false
      eventbridge_rules: true
      eventbridge_pipes: true
      s3_notifications: false
    # Match the s3 source recipe, for S3 notification lineage.
    # platform_map:
    #   s3:
    #     platform_instance: "data-lake"
    #     env: "PROD"
    #     convert_urns_to_lowercase: false

    cloudtrail_producers:
      enabled: false
      # event_data_store: "arn:aws:cloudtrail:us-east-1:123456789012:eventdatastore/EXAMPLE"
      # lookback_days: 7

    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
include_column_lineage
One of boolean, null
Add field-level lineage on hops that deliver the message unchanged or with a simple field mapping, for fields present in both schemas. See the docs for which hops qualify. Schemas come from Glue Schema Registry in this run or from DataHub, one DataHub lookup per hop for a schema not read from Glue. Defaults to glue_schema_registry.enabled; set it to true to use schemas that are already in DataHub without Glue.
Default: None
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
cloudtrail_producers
CloudTrailProducersConfig
cloudtrail_producers.enabled
boolean
Find producers from CloudTrail Lake: who called SendMessage or Publish on this source's queues or topics. Needs data-event logging for SQS/SNS into an event data store. Lake queries are billed per GB scanned.
Default: False
cloudtrail_producers.event_data_store
One of string, null
ARN or id of the CloudTrail Lake event data store to query. With an ARN the query runs in the store's region; with a bare id, in this recipe's region. A store in another account must be shared with this role through its resource policy.
Default: None
cloudtrail_producers.lookback_days
integer
How many days of events the producer query covers.
Default: 7
cloudtrail_producers.query_timeout_seconds
integer
How long to wait for the CloudTrail Lake query before giving up.
Default: 300
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
glue_schema_registry
QueueGlueSchemaRegistryConfig
glue_schema_registry.enabled
boolean
Attach schemas from AWS Glue Schema Registry. Needs glue:GetSchemaVersion. AWS stores no link between a queue or topic and a schema, so list them in the schema map.
Default: False
glue_schema_registry.queue_schema_map
map(str,string)
glue_schema_registry.registry_name
string
Glue Schema Registry registry to read schemas from.
Default: default-registry
glue_schema_registry.use_naming_convention
boolean
When the schema map has no entry, look for a schema with the same name as the queue or topic. Off by default: AWS defines no such convention.
Default: False
lineage
AwsMessagingLineageConfig
lineage.eventbridge_pipes
boolean
Emit EventBridge Pipes whose source or target is one of this source's queues or topics.
Default: True
lineage.eventbridge_rules
boolean
Emit EventBridge rules that target this source's queues or topics.
Default: True
lineage.lambda_async_destinations
boolean
Read each function's asynchronous invocation destinations. One lambda:ListFunctionEventInvokeConfigs call per function in the region, made by both the sqs and sns sources; stops at the first AccessDenied. Stale Lambda output edges are only removed when this is on, since without it the run can't tell which outputs still exist.
Default: False
lineage.lambda_functions
boolean
Emit Lambda functions wired to this source's queues or topics (event source mappings, SNS subscriptions, failure and async destinations) as a DataFlow with one DataJob.
Default: True
lineage.s3_notifications
boolean
Add S3 datasets as upstreams of the queues or topics their bucket's event notifications target: every S3 dataset already in DataHub under the notification's bucket and key prefix, up to 100. Reads the notification configuration of each bucket in the recipe's region (one call per bucket); a source only links its own queues or topics. Needs a DataHub graph connection, and s3:ListAllMyBuckets plus s3:GetBucketNotification. ListAllMyBuckets only returns buckets the recipe's account owns.
Default: False
platform_map
map(str,LinkedPlatformDetail)
platform_map.key.platform_instance
One of string, null
platform_instance of that platform's own recipe. If unset, the dataset URN has no platform_instance.
Default: None
platform_map.key.convert_urns_to_lowercase
boolean
Set to that platform's recipe convert_urns_to_lowercase. Defaults to False, like the s3 source.
Default: False
platform_map.key.env
One of string, null
env of that platform's own recipe. If unset, inherits this recipe's env.
Default: None
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.

Schemas from Glue Schema Registry​

SQS stores opaque messages, so there is no schema by default. If producers register message schemas in AWS Glue Schema Registry, set glue_schema_registry.enabled and list each queue and its schema in glue_schema_registry.queue_schema_map. AWS keeps no link between a queue and a schema, so the map is the reliable way; use_naming_convention looks for a schema named like the queue when the map has no entry. AVRO, JSON and PROTOBUF schemas are supported, read with glue:GetSchemaVersion. PROTOBUF also needs the kafka extra, for example pip install 'acryl-datahub[sqs,kafka]'.

Lambda and EventBridge​

With the lineage options on (the default), the connector follows the services around its queues:

EdgeRead from
Queue to Lambda functionlambda:ListEventSourceMappings
Lambda function to queueevent source mapping OnFailure destination; asynchronous invocation destinations (lineage.lambda_async_destinations, off by default)
Event bus (none for a scheduled rule) to rule to queueevents:ListRules, events:ListTargetsByRule
Pipe from or to a queuepipes:ListPipes
  • A Lambda function is a DataFlow (platform lambda) with one invoke DataJob, the shape the kinesis source uses for Firehose. Functions appear only when one of these edges involves one of the run's queues, and only for the recipe's own account and region.
  • A rule (platform eventbridge) is a DataFlow with one route DataJob from its event bus, emitted as an Event Bus dataset, to its targets. A scheduled rule has no event bus input, so its job has outputs only. A pipe (platform eventbridge-pipes) is a DataFlow with one pipe DataJob from source to target. A rule or pipe that targets a function becomes an upstream job of that function.
  • The sqs and sns sources each add only the edges whose dataset end they own, as patches, so a rule targeting both a queue and a topic gets both outputs once both have run.
  • Some pipes also get field-level lineage; see the table below.

When column lineage is emitted​

Field-level lineage is only added where a message reaches the next hop unchanged (or, for a pipe, where its input template maps body fields to keys), for fields present in both schemas (matched case-insensitively on the simple field path). Schemas come from Glue Schema Registry in the same run or from DataHub. Without a schema on both sides, the edge stays table-level. Most deployments have no schema on one side and see table-level lineage only. The report counts the reason for each hop it skips under column_lineage_skipped; with include_column_lineage off, nothing is attempted or counted, and existing field edges are left alone.

HopColumn lineageWhy
Queue to its dead-letter queue (sqs)Yes, field to same fieldSQS moves the failed message itself
Topic to queue, raw message delivery (sns, with include_subscription_attributes)Yes, field to same fieldThe queue receives the published message as is
Topic to queue, without raw delivery (sns)NoSNS wraps the message in its JSON envelope
Topic to a subscription's dead-letter queue (sns)NoAWS doesn't document whether that queue holds the raw message or the envelope
Pipe from a queue whose input template only has <$.body.path> valuesYes, each template key from its body pathPipes decodes the SQS body and fills each key from one field
Pipe without an input template, with an enrichment step, or with literals or concatenation in the templateNoWithout a template the target gets the SQS record (message id, attributes, body), not the message
Any hop through a Lambda functionNoThe function can change anything

include_column_lineage defaults to glue_schema_registry.enabled. With it on, each queue with a dead-letter queue costs a DataHub lookup for any schema not read from Glue this run. Set it to true without Glue to use schemas already in DataHub, or false to turn it off.

Field edges are retracted along with table-level ones (remove_stale_lineage): when a field is renamed or dropped, field edges from that upstream that this run no longer emits are removed. A hop whose field lineage was skipped this run (no schema on one side, a Glue schema that couldn't be read, or include_column_lineage off) keeps its field edges.

S3 event notifications​

lineage.s3_notifications (off by default) reads the notification configuration of each bucket in the recipe's region and adds the matching S3 datasets as upstreams of the queues each notification targets; the bucket itself is not linked. The s3 source names datasets after a path_spec's table folder (landing/incoming/orders for s3://landing/incoming/{table}/*.parquet), while a notification only has a key prefix filter such as incoming/. A prefix ending in / matches the dataset <bucket>/incoming and every dataset under <bucket>/incoming/; any other prefix matches dataset names that start with <bucket>/<prefix>, and no prefix matches every dataset in the bucket. A suffix filter such as .csv is checked against datasets named after a file the s3 source reads (csv, tsv, json, jsonl, parquet or avro, optionally compressed, such as <bucket>/incoming/2024.csv.gz); datasets named after a folder, including dotted ones such as orders.v2, still match, since their files aren't known. Up to 100 datasets per notification are linked, the first by name (a warning says when that cap is hit). No S3 datasets are created, and soft-deleted ones are not linked. The lookup needs a DataHub graph connection (the datahub-rest sink or datahub_api); without one no S3 upstreams are added and the report lists the skipped edges. Run the s3 source first and set platform_map.s3 (platform_instance, env, convert_urns_to_lowercase) to match that recipe. While a notification matches no S3 dataset in DataHub, earlier S3 edges into its queue are kept. Only edges from buckets whose notifications this run read are removed. s3:ListAllMyBuckets only returns buckets the recipe's account owns, so an edge from a bucket in another account is kept. This makes one s3:GetBucketNotification call per bucket in the recipe's region. Filtering buckets by region needs botocore 1.35.42 or later; with an older botocore the connector warns and reads every bucket's notification configuration.

Each source only links its own queues, so turn this on in the sqs recipe if buckets notify queues and in the sns recipe if they notify topics, including when one bucket notifies both. Each source reads every bucket's notification configuration. When the bucket listing succeeds, an S3 edge whose notification was removed is removed on the next run (remove_stale_lineage) if its bucket's notifications were read; a bucket whose notifications couldn't be read keeps its edges, and a failed listing removes none; this reads the current lineage of each of the source's queues from DataHub.

Producers from CloudTrail Lake​

Code that sends with the AWS SDK leaves no trace in any configuration API, so most queues have no upstream. If you log SQS and SNS data events to a CloudTrail Lake event data store, set cloudtrail_producers.enabled and cloudtrail_producers.event_data_store. The connector runs one query per run over the last lookback_days of SendMessage and SendMessageBatch events and groups them by caller and queue.

  • With lineage.lambda_functions on, a caller that is a Lambda function in the recipe's account and region becomes an upstream job of the queue. The match relies on Lambda naming the execution role session after the function, and the caller's role must be that function's execution role. That convention has not yet been checked against a real trail for every runtime and invocation path, so treat these edges as best effort.
  • Any other caller (an ECS task or EC2 role, an IAM user) is listed by caller ARN (the role ARN for an assumed role, the user ARN for an IAM user) in the queue's observed_producers property, not drawn as lineage, because a role or user is not a DataHub entity.
  • The query only covers events in the recipe's region, and SQS targets are matched on the region and account in each queue URL (or queue ARN, which SendMessage also accepts), so a multi-region trail doesn't mix up same-named queues.
  • Give event_data_store as an ARN when the store is in another region or another account: the query runs in the store's region, and the owner's account id comes from the ARN. A store in another account must also be shared with the ingestion role through its resource policy.
  • If the query fails, the queue's existing observed_producers value is read back from DataHub and kept, which needs a DataHub connection (the datahub-rest sink provides one). Without one it is cleared until the next successful run, with a warning. If the query succeeds and no role sent to the queue in the lookback window, the value is removed. A Lambda producer edge that the window no longer shows is removed with the other stale Lambda output edges (remove_stale_lineage, with lineage.lambda_async_destinations on); a failed query keeps them.
  • CloudTrail Lake bills queries by data scanned, and data events must be enabled for the queues, which also costs. Both are off by default. Each run scans the whole lookback_days window (default 7), so a daily schedule scans each day about seven times; lower lookback_days to match the schedule if cost matters. If the query fails, each queue also costs one DataHub read of its datasetProperties to keep observed_producers.

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.

  • Schema metadata comes from Glue Schema Registry (queue_schema_map or use_naming_convention) or from schemas already in DataHub; SQS itself stores opaque message bodies.

  • Queue depth (ApproximateNumberOfMessages and similar) is not ingested; it changes every few seconds.

  • Lambda functions, rules, pipes and event buses can be emitted by both the sqs and sns sources, so neither source soft-deletes them through stale-entity removal: one source dropping a function it no longer references would delete it while the other still does. A function, rule or pipe deleted in AWS stays in DataHub until removed there. Its edges to this source's queues are still removed on the next run (remove_stale_lineage), so it stops showing up in their lineage. Edges are only removed for a kind of resource that was read in full: a failed or turned-off listing keeps them, and Lambda output edges are only removed with lineage.lambda_async_destinations on, since some outputs come from those destinations. Only jobs in the recipe's own account and region are checked: an edge another account's recipe wrote is left for that recipe. A pipe's field edges are removed with its output edge, or when its input template no longer maps that field.

  • The lambda, eventbridge and eventbridge-pipes region containers are shared with the other source in the same way.

  • Producers that send from application code are not visible in these APIs, so many queues and topics have no upstream.

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.