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
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.
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❌SNS has no schema.
Table-Level Lineage✅Topic to subscribed queue and to subscription dead-letter queue.
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, so topics carry no schema fields. 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": "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.

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"

    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_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
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
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.

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.
  • No schema metadata: SNS 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 Lambda, Firehose and HTTP endpoints are counted but not yet drawn as lineage.

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.