DEV Community: Falk

Snowflake Task Error Notifications with Terraform/ AWS SNS

Falk — Wed, 17 Aug 2022 12:56:41 +0000

Snowflake Tasks are a great way to automatically execute SQL queries on a schedule. In this guide, we'll focus on setting up notifications which will be sent to an email address when a task run fails. As in Snowflakes Official Documentation, we'll use AWS SNS for this.

Terraform Setup
Create SNS Topic and IAM Policy
Creating the Snowflake Notification Integration
Granting Snowflake Access to the SNS Topic
Enabling Error Notifications in Tasks
Receive Email Notifications

Terraform Setup

First, we need to create the terraform folder structure:

mkdir terraform
touch terraform/variables.tf
touch terraform/providers.tf
touch terraform/outputs.tf
touch terraform/main.tf

`variables.tf`

variable "name" {
  description = "the name of your stack, e.g. \"demo\""
  default     = "snowflake-error-notifications"
}

variable "environment" {
  description = "the name of your environment, e.g. \"prod\""
  default     = "prod"
}

variable "region" {
  description = "the AWS region in which resources are created, you must set the availability_zones variable as well if you define this value to something other than the default"
  default     = "eu-central-1"
}

`providers.tf`

provider "aws" {
  region = var.region
}

`outputs.tf`

output "sns_topic_arn" {
  value = "${aws_sns_topic.snowflake-error-notifications.arn}"
}

output "iam_role_arn" {
  value = "${aws_iam_role.snowflake-error-notifications-role.arn}"
}

Create SNS Topic and IAM Policy

`main.tf`

resource "aws_sns_topic" "snowflake-error-notifications" {
  name = "sns-${var.name}-${var.environment}"

  tags = {
    Name = "${var.name}"
    Env = "${var.environment}"
  }
}

resource "aws_iam_policy" "snowflake-error-notifications-policy" {
  name        = "policy-${var.name}-${var.environment}"
  description = "Policy for Snowflake Tasks error notifications"

  policy = jsonencode({
    "Version": "2012-10-17",
    "Statement": [
      {
         "Effect": "Allow",
         "Action": [
             "sns:Publish"
         ],
         "Resource": "${aws_sns_topic.snowflake-error-notifications.arn}"
      }
    ]
 })

  tags = {
    Name = "${var.name}"
    Env = "${var.environment}"
  }
}

resource "aws_iam_role" "snowflake-error-notifications-role" {
  name = "role-${var.name}-${var.environment}"
  assume_role_policy = jsonencode({
    "Version": "2012-10-17",
    "Statement": [
      {
        "Effect": "Allow",
        "Principal": {
          "Service": "sns.amazonaws.com"
        },
        "Action": "sts:AssumeRole"
      }
    ]
  })

  tags = {
    Name = "${var.name}"
    Env = "${var.environment}"
  }
}

resource "aws_iam_policy_attachment" "snowflake-error-notifications-policy-attachment" {
  name       = "policy-attachment-${var.name}-${var.environment}"
  roles      = ["${aws_iam_role.snowflake-error-notifications-role.name}"]
  policy_arn = "${aws_iam_policy.snowflake-error-notifications-policy.arn}"
}

Now we can create the terraform configuration for our stack:

cd terraform
terraform init
terraform apply

Creating the Snowflake Notification Integration

In Snowflake, we'll create the notification integration. Please note that you need to be ACCOUNTADMIN for this to work.
Get SNS topic and IAM Role ARN from the Terraform output and add it to:

create notification integration task_error_notifications
  enabled = true
  type = queue
  notification_provider = aws_sns
  direction = outbound
  aws_sns_topic_arn = 'arn:aws:sns:XXX:sns-snowflake-error-notifications-prod'
  aws_sns_role_arn = 'arn:aws:iam::XXX:role/role-snowflake-error-notifications-prod';

Granting Snowflake Access to the SNS Topic

desc notification integration task_error_notifications;

You will need SF_AWS_IAM_USER_ARN and SF_AWS_EXTERNAL_ID from the query output.
Next, head back to our terraform code and add these values to the aws_iam_role resource:

resource "aws_iam_role" "snowflake-error-notifications-role" {
  name = "role-${var.name}-${var.environment}"
  assume_role_policy = jsonencode({
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "",
      "Effect": "Allow",
      "Principal": {
        "AWS": "<sf_aws_iam_user_arn>"
      },
      "Action": "sts:AssumeRole",
      "Condition": {
        "StringEquals": {
          "sts:ExternalId": "<sf_aws_external_id>"
        }
      }
    }
  ]
})

  tags = {
    Name = "${var.name}"
    Env = "${var.environment}"
  }
}

Now update the terrafom configuration:

terraform apply

Enabling Error Notifications in Tasks

The only thing left is to enable the integration we just created in our Snowflake Tasks.

For new tasks:

create task new_task
  schedule = '5 MINUTE'
  error_integration = task_error_notifications
  as
  insert into mytable(ts) values(current_timestamp);

For existing tasks:

alter task old_task suspend;
alter task old_task set error_integration = task_error_notifications;
alter task old_task resume;

In case a task fails, the payload will be sent to the SNS topic as a JSON string:

{\"version\":\"1.0\",\"messageId\":\"3ff1eff0-7ad7-493c-9552-c0307087e0c6\",\"messageType\":\"USER_TASK_FAILED\",\"timestamp\":\"2021-11-11T19:46:39.648Z\",\"accountName\":\"AWS_UTEN_DPO_ACC\",\"taskName\":\"AWS_UTEN_DPO_DB.AWS_UTEN_SC.UTEN_AWS_TK1\",\"taskId\":\"01a03962-2b57-889e-0000-000000000001\",\"rootTaskName\":\"AWS_UTEN_DPO_DB.AWS_UTEN_SC.UTEN_AWS_TK1\",\"rootTaskId\":\"01a03962-2b57-889e-0000-000000000001\",\"messages\":[{\"runId\":\"2021-11-11T19:46:23.826Z\",\"scheduledTime\":\"2021-11-11T19:46:23.826Z\",\"queryStartTime\":\"2021-11-11T19:46:24.879Z\",\"completedTime\":\"null\",\"queryId\":\"01a03962-0300-0002-0000-0000000034d8\",\"errorCode\":\"000630\",\"errorMessage\":\"Statement reached its statement or warehouse timeout of 10 second(s) and was canceled.\"}]}

Receive Email Notifications

If you'd like to receive an email every time a task fails, you can add the following to main.tf:

resource "aws_sns_topic_subscription" "snowflake-error-notifications-subscription" {
  topic_arn = "${aws_sns_topic.snowflake-error-notifications.arn}"
  protocol  = "email"
  endpoint  = "youremail@example.com"
}

And apply the changes:

terraform apply

You will receive an Email from AWS where you have to confirm the subscription:

Create Jira Ticket on Prefect Task Failure

Falk — Tue, 16 Aug 2022 13:22:20 +0000

Prefect offers a great UI to monitor and manage your Prefect runs. And there is nothing better than opening your laptop in the morning and seeing all green when you look at it. However, that's unfortunately not always the case and, even though this depends on your organization's workflows, usually you will have to create a bug ticket in Jira before attempting to resolve any issues in your runs. Let's automate this by integrating the jira_notifier in our tasks.

Prerequisites

First, we'll need to create a Jira API token. You can do this by following this URL, logging in, and then click the Create API token button. Write down the token and store it somewhere safe, we'll need it in a second.

Next we'll have to pass the token we just created, the Jira user and the Jira server URL on to Prefect. One easy way to do this is to add the following to your Prefect configuration file (~/.prefect/config.toml):

[context.secrets]
JIRASECRETS.JIRATOKEN = "thetokenyoucreated"
JIRASECRETS.JIRAUSER = "xxx@example.com"
JIRASECRETS.JIRASERVER = "https://xxx.atlassian.net"

Configure the Prefect Task

Now we can add the jira_notifier to our Prefect task:

from prefect import Task
from prefect.utilities.notifications.jira_notification import jira_notifier
from prefect.engine.state import Failed

@task(
    name="jira_example_task",
    state_handlers=[
        jira_notifier(
            only_states=[Failed],
            options={"project": "DE", "issuetype": {"name": "Bug"}},
        )
    ],
)
def add(x, y):
    return x + y

This example task will create a Jira Bug ticket in the DE project when the run fails.
One thing to keep in mind is that the issue type you're selecting must be available in the project you're creating the issue in. You can also change the state to, for example, Retrying to create a Jira Task ticket instead. An overview of the various states is available here.

The created ticket will look like this:

Hash Personal Identifiable Information (PII) in your ELT pipelines

Falk — Mon, 15 Aug 2022 12:38:13 +0000

What is Personal Identifiable Information?

Personal Identifiable Information (PII) is defined as: Any representation of information that permits the identity of an individual to whom the information applies to be reasonably inferred by either direct or indirect means. If you collect, use or store PII of people in the European Union, you have to work GDPR-compliant and therefore should protect your customers personal data.

So what, I just delete those

While you could of course think about deleting any PII from your data before adding it to your file storage/ database, the data users in your organization might have a word to say about that. Oftentimes, Analysts or Data Scientists have to work with multiple data sources and find connections between them in order to generate insights.
For example, finding a customer across various sales platforms such as Amazon, Shopify, or eBay. These platforms don't have a common unique identifier for each user, so an alternative such as an email, name, or phone number must be used. That is however not possible anymore if we've decided to delete this personal information. So let's try to add a "Hashing" step to our ELT pipeline.

Introducing ByeByePii

ByeByePii is a Python package that is meant for hashing personal identifiable information. It was built focused on making Data Lakes storing JSON files GDPR-compliant. It's a simple package with two features:

Analyzing Python Dictionaries in order to identify PII
Hashing PII in a given Python dictionary

Binary installers for the latest released version are available at the Python Package Index (PyPI):

pip install ByeByePii

Analyzing a JSON and creating a list of keys to hash

In order to not having to manually look for all the keys in a Python Dictionary, we can use the analyzeDict function.

import byebyepii
import json

if __name__ == '__main__':

    # Loading local JSON file
    with open('data.json') as json_file:
        data = json.load(json_file)

    # Analyzing the dictionary and creating our hash list
    key_list, subkey_list = byebyepii.analyzeDict(data)

$ python3 analyzeDict.py

Add BuyerInfo - BuyerEmail to hash list? (y/n) y
Add SalesChannel to hash list? (y/n) n
Add OrderStatus to hash list? (y/n) n
Add PurchaseDate to hash list? (y/n) n
Add ShippingAddress - StateOrRegion to hash list? (y/n) y
Add ShippingAddress - PostalCode to hash list? (y/n) y
Add ShippingAddress - City to hash list? (y/n) n
Add ShippingAddress - CountryCode to hash list? (y/n) n
Add LastUpdateDate to hash list? (y/n) n

Keys to hash: ['BuyerInfo', 'ShippingAddress', 'ShippingAddress', 'ShippingAddress', 'ShippingAddress']
Subkeys to hash: ['BuyerEmail', 'StateOrRegion', 'PostalCode']

Hashing PII in a given JSON

Using the key lists we just created we can proceed to hash the PII in the dictionary.

import byebyepii
import json

if __name__ == '__main__':

    # Loading local JSON file
    with open('data.json') as json_file:
        data = json.load(json_file)

    # Hasing the PII
    keys_to_hash = ['BuyerInfo', 'ShippingAddress', 'ShippingAddress', 'ShippingAddress', 'ShippingAddress']
    subkeys_to_hash = ['BuyerEmail', 'StateOrRegion', 'PostalCode']
    hashed_pii = byebyepii.hashPii(data, keys_to_hash, subkeys_to_hash)

    # Writing the updated JSON file
    with open('hashed_data.json', 'w') as outfile:
        json.dump(hashed_pii, outfile)

Before:

{
  "BuyerInfo": {
    "BuyerEmail": "test@test.com"
  },
  "EarliestShipDate": "2022-01-01T23:59:59Z",
  "SalesChannel": "Website",
  "OrderStatus": "Shipped",
  "PurchaseDate": "2022-01-01T23:59:59Z",
  "ShippingAddress": {
    "StateOrRegion": "West Midlands",
    "PostalCode": "DY9 0TH",
    "City": "STOURBRIDGE",
    "CountryCode": "GB"
  },
  "LastUpdateDate": "2022-01-01T23:59:59Z",
}

After:

{
  "BuyerInfo": {
    "BuyerEmail": "037a51cb9162f51772eaf6b0fb02e1b5d0bf8219deacf723eeedc162209bfd33"
  },
  "EarliestShipDate": "2022-01-01T23:59:59Z",
  "SalesChannel": "Website",
  "OrderStatus": "Shipped",
  "PurchaseDate": "2022-01-01T23:59:59Z",
  "ShippingAddress": {
    "StateOrRegion": "08fa57d00de1936ebea7aeaf8e36d04510a5d885cfaa4f169c2b010d36ccaca4",
    "PostalCode": "714f02c01e20988ee273776dc218f44326c2f5839618b0c117413b0cc7d91701",
    "City": "STOURBRIDGE",
    "CountryCode": "GB"
  },
  "LastUpdateDate": "2022-01-01T23:59:59Z",
}

Since the string test@test.com will always be hashed to 037a51cb9162f51772eaf6b0fb02e1b5d0bf8219deacf723eeedc162209bfd33 it is still perfectly usable as a cross-functional identifier.