Skip to main content

Amazon SNS

Overview​

Amazon SNS is AWS's managed pub/sub service. Publishers send messages to a topic and SNS delivers a copy to each subscriber: SQS queues, Lambda functions, Firehose streams, HTTP endpoints, email or SMS.

The sns source plugin catalogs the topics of one AWS region as datasets, grouped in a region container, with topic settings and subscription counts as properties, AWS resource tags as DataHub tags, lineage from each topic to its subscribed SQS queues and subscription dead-letter queues, optional tags or glossary terms from message data protection policies, and stateful deletion. Its companion, the sqs source, catalogs the queues; 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 topic ARN
AWS regionContainerSubtype Region
TopicDatasetPlatform sns, subtype Topic; name <region>.<topic name>
SQS subscription, redrive policyUpstream lineage (patch) on the queueTopic to queue, topic to subscription dead-letter queue
Lambda function wired to a topicDataFlow + DataJobPlatform lambda; one invoke job per function
EventBridge rule or Pipe touching a topicDataFlow + DataJobPlatforms eventbridge and eventbridge-pipes; event buses are Event Bus datasets
Resource tagTagKey:Value, or Key for an empty value
Data protection identifierTag or Glossary TermOpt-in; terms only for identifiers you map

Module sns​

Alpha

Important Capabilities​

CapabilityStatusNotes
Asset Containers✅Region containers.
Column-level Lineage✅Opt-in: topic to raw-delivery queue, when both have schemas.
Descriptions❌SNS 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.
Glossary Terms✅Opt-in, from data protection policy identifiers you map to existing terms.
Platform Instance✅Defaults to the AWS account id.
Schema Metadata✅Opt-in: AWS Glue Schema Registry schemas listed in glue_schema_registry.topic_schema_map, or found by use_naming_convention.
Table-Level Lineage✅Subscribed queues and dead-letter queues, Lambda, EventBridge rules and Pipes.
Test Connection✅Enabled by default.

Overview​

The sns module reads the SNS control-plane APIs for one AWS region per recipe. It never publishes or reads messages.

SNS has no schema of its own; topic schemas can be attached from AWS Glue Schema Registry (opt-in). Topic settings (display name, FIFO, KMS key, archive policy) and subscription counts per protocol are stored as custom properties.

SQS queues are ingested by the separate sqs source. Topic-to-queue lineage is written by this source onto the queue URNs, so run both with the same env, platform_instance and account_platform_instance_map, or the edges will point at queue URNs the sqs source doesn't emit.

Prerequisites​

AWS IAM Permissions​

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "SnsRead",
      "Effect": "Allow",
      "Action": [
        "sns:ListTopics",
        "sns:GetTopicAttributes",
        "sns:ListTagsForResource",
        "sns:ListSubscriptionsByTopic",
        "sns:GetSubscriptionAttributes",
        "sns:GetDataProtectionPolicy"
      ],
      "Resource": "*"
    },
    {
      "Sid": "QueueExistence",
      "Effect": "Allow",
      "Action": ["sqs:ListQueues"],
      "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": "*"
    }
  ]
}
  • sqs:ListQueues tells the connector which queues in its own account and region exist, so a subscription to a deleted queue doesn't create an empty queue entity. Without it, edges to local queues are skipped with a warning.
  • Tags are read with one tag:GetResources call. If it is denied the connector warns and calls sns:ListTagsForResource per topic, which SNS limits to 10 calls per second.
  • Subscriptions are listed per topic with sns:ListSubscriptionsByTopic, because sns:ListSubscriptions only returns the caller's own subscriptions and would miss ones other accounts created on your topics. That is one call per topic, and SNS allows 30 per second.
  • sns:GetSubscriptionAttributes is only called when include_subscription_attributes is on, and sns:GetDataProtectionPolicy only when data_protection.enabled is on.

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[sns]'

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: sns
  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 as the sqs recipe so topic-to-queue lineage joins.
    env: "PROD"

    topic_pattern:
      allow:
        - ".*"

    extract_tags: true
    # One sns:GetSubscriptionAttributes call per subscription; reads subscription DLQs.
    include_subscription_attributes: false

    data_protection:
      enabled: false
      # glossary_term_map:
      #   EmailAddress: "urn:li:glossaryTerm:pii.email"

    # account_platform_instance_map:
    #   "210987654321": "shared-services"

    glue_schema_registry:
      enabled: false
      registry_name: "default-registry"
      # Set enabled: true and replace {} with a mapping, for example:
      #   topic_schema_map:
      #     orders: "order-v1"
      topic_schema_map: {}

    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
include_subscription_attributes
boolean
Call sns:GetSubscriptionAttributes for each subscription to read its dead-letter queue. One call per subscription, for every protocol, on every run; subscription-to-queue edges need no extra calls.
Default: False
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
data_protection
SnsDataProtectionConfig
data_protection.enabled
boolean
Read each topic's message data protection policy (sns:GetDataProtectionPolicy, one call per topic) and tag the topic with the data identifiers it covers.
Default: False
data_protection.glossary_term_map
map(str,string)
data_protection.tag_prefix
string
Tags are named <tag_prefix>:<identifier>, e.g. aws_data_identifier:EmailAddress.
Default: aws_data_identifier
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
TopicGlueSchemaRegistryConfig
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.registry_name
string
Glue Schema Registry registry to read schemas from.
Default: default-registry
glue_schema_registry.topic_schema_map
map(str,string)
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
topic_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$".
topic_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 topic ARN and uses it. Dataset names include the region. The sqs source names queues the same way, through the same shared code.

Subscription lineage​

EdgeRead from
Topic to SQS queuesns:ListSubscriptionsByTopic, protocol sqs
Topic to a subscription's dead-letter queuethe subscription's RedrivePolicy (include_subscription_attributes)

Edges are upstream lineage patches on the queue, so the queue keeps its dead-letter lineage from the sqs source and edges from other accounts' recipes. Each edge carries a relationship property (subscription or subscription_dead_letter).

When a subscription is removed, the next run removes its edge (remove_stale_lineage, on by default). This reads each topic's current lineage from DataHub, so it needs a graph connection (the datahub-rest sink or datahub_api); without one, old edges stay. Only subscription edges are removed, plus subscription_dead_letter edges when include_subscription_attributes is on, and only from topics whose subscriptions were all read this run. If sqs:ListQueues is denied or fails partway, no edges are removed that run. Edges the sqs source or other accounts' recipes write are never touched.

include_subscription_attributes is off by default: it makes one sns:GetSubscriptionAttributes call per subscription, of every protocol, on every run. The count is in the report as subscription_attribute_calls. Turn it on if subscriptions have dead-letter queues you want in lineage.

A queue in the recipe's own account and region gets an edge only if sqs:ListQueues shows it exists. A queue in another account or region gets an edge under that account's platform instance: the account id, or the value in account_platform_instance_map. Unconfirmed and deleted subscriptions, which SNS lists with a placeholder such as PendingConfirmation or Deleted instead of an ARN, are skipped. Skipped edges are listed in the report under lineage_edges_skipped.

Topics record how many subscriptions each protocol has, and the hosts of HTTP(S) endpoints (web_endpoint_hosts). Email and SMS endpoints and full endpoint URLs are never stored.

Schemas from Glue Schema Registry​

SNS 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 topic and its schema in glue_schema_registry.topic_schema_map. AWS keeps no link between a topic and a schema, so the map is the reliable way; use_naming_convention looks for a schema named like the topic 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[sns,kafka]'.

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 each skip reason under column_lineage_skipped.

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)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 and needs include_subscription_attributes to see raw delivery. With it on, each distinct raw-delivery topic-to-queue hop may cost a DataHub lookup for a schema not read from Glue this run; repeated hops are added once and lookups are cached. 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, or a subscription turns raw delivery off, 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, include_subscription_attributes off, or include_column_lineage off) keeps its field edges.

Lambda and EventBridge​

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

EdgeRead from
Topic to Lambda functionSNS subscription, protocol lambda
Lambda function to topicevent source mapping OnFailure destination; asynchronous invocation destinations (lineage.lambda_async_destinations, off by default)
Event bus (none for a scheduled rule) to rule to topicevents:ListRules, events:ListTargetsByRule
Pipe from or to a topicpipes: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 topics, 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 above.

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 topics 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 topic 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 topics, 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 topics from DataHub.

Producers from CloudTrail Lake​

Code that sends with the AWS SDK leaves no trace in any configuration API, so most topics 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 Publish and PublishBatch events and groups them by caller and topic.

  • 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 topic. 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 topic'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 SNS targets are matched by topic ARN, so a multi-region trail doesn't mix up same-named topics.
  • 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 topic'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 topic 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 topics, 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 topic also costs one DataHub read of its datasetProperties to keep observed_producers.

Tags and data protection​

AWS resource tags become DataHub tags named Key:Value (or Key for an empty value), the same format as the kinesis and sqs sources. They are added to the topic rather than replacing its tags, so tags added in the DataHub UI stay; a tag removed in AWS stays in DataHub until removed there. Use the extract_ownership_from_tags transformer to turn an owner tag into ownership.

With data_protection.enabled, each topic's message data protection policy is read and every data identifier it names (for example EmailAddress) becomes a tag aws_data_identifier:EmailAddress. Map identifiers to existing glossary terms with data_protection.glossary_term_map to attach terms as well; no terms are created. These tags and terms are added the same way as resource tags: if a topic's policy stops covering an identifier, its tag and term stay until removed in DataHub, so check them before relying on them as a current classification.

Domains​

domain maps a domain urn or id to regex patterns on topic names, as in the glue source.

Limitations​

  • One region per recipe.
  • Schema metadata comes from Glue Schema Registry (topic_schema_map or use_naming_convention) or from schemas already in DataHub; SNS itself doesn't store messages.
  • Lineage to a queue that the sqs recipe's queue_pattern excludes is still written, because the queue exists and its URN follows from its ARN. This source doesn't read the sqs recipe's filters. The queue then appears in DataHub with only its lineage.
  • Subscriptions to HTTP, email and SMS endpoints are counted but not drawn as lineage.
  • 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 topics 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 on sns:ListTopics: add the statement above.
  • "Could not read a resource ARN to find the AWS account id" error: the first sns:ListTopics call failed, so the run stopped rather than write topics under URNs without the account id. Re-run, or set platform_instance explicitly.
  • Edges skipped with "not found in this region": the queue doesn't exist, or sqs:ListQueues was denied.
  • Lineage doesn't join with the sqs source: check that both recipes use the same env, platform_instance and account_platform_instance_map.

Code Coordinates​

  • Class Name: datahub.ingestion.source.aws_messaging.sns_source.SnsSource
  • Browse on GitHub
Questions?

If you've got any questions on configuring ingestion for Amazon SNS, 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.