Introduction

Building on our previous blog Building ETL Pipeline in Snowflake, we will cover different approaches to automated testing for ETL Pipelines in Snowflake. For these kinds of solutions, QA Engineers should use their creativity to implement useful test cases as much as possible. There are  various approaches, and we will go through the following: tasks verification, end-to-end testing, IQR analysis and data quality verification.

Tasks verification

This part of the automated testing helps the verification of E and T of ETL parts. If we examine the example from the previously mentioned blog, we can see the use of tasks in Snowflake ETL pipelines. This allows us to develop an automated test case that runs a Snowflake query to identify failed tasks within the last 24 hours or a specified time range, depending on your needs. Here is the example written in Node JS with Jest (or any other test runner that you prefer, such as Jasmine, etc.):

1. Initialise repository and install Jest with commands:

npm init
npm i jest

2. Create Snowflake Connector Util class:

const Snowflake = require('snowflake-promise').Snowflake

const connectionProperties = {
   account: 'TEST_ACCOUNT',
   username: 'TEST_USERNAME',
   role: 'TEST_ROLE',
   database: 'TEST_DB',
   schema: 'TEST_SCHEMA',
   warehouse: 'TEST_WAREHOUSE',
   authenticator: 'SNOWFLAKE_JWT',
   privateKeyPath: process.env.SF_TEST_PRIVATE_KEY,
   privateKeyPass: process.env.SF_TEST_PRIVATE_KEY_PASS,
}

const taskHistoryTable = `${connectionProperties.database}.INFORMATION_SCHEMA.TASK_HISTORY`

class SnowflakeConnectorUtil {

   async snowflakePool() {
       return new Snowflake(connectionProperties)
   }

   async createPool() {
       const pool = await this.snowflakePool()
       await pool.connect()

       return pool
   }

   async getSnowflakeFailedTasksInLast24Hours() {
       return `SELECT *
      FROM (
         SELECT
           distinct name,
           first_value(completed_time) ignore nulls over (partition by name order by completed_time desc) as last_completed_time,
           first_value(state) ignore nulls over (partition by name order by completed_time desc) as last_state
         FROM db.schema.task_history
       )
       WHERE last_state IN ('FAILED', 'CANCELLED', 'FAILED_AND_AUTO_SUSPENDED') AND last_completed_time >= DATEADD(DAY, -1, GETDATE());`
   }
}

module.exports = SnowflakeConnectorUtil

3. Create a test file. For your test case use previously created Snowflake Connector Util:

const SnowflakeConnectorUtil = require('./db/SnowflakeConnectorUtil')
const snowflakeConnectorUtil = new SnowflakeConnectorUtil()

describe('Tasks verification', () => {
   let pool

   beforeAll(async () => {
       pool = await snowflakeConnectorUtil.createPool()
   }, 10000)

   it('Daily verification tasks', async () => {
       const query = await snowflakeConnectorUtil.getSnowflakeFailedTasksInLast24Hours()
       const records = await pool.execute(query)
       records.map(failedTask => {
           console.log(
               `Failed task + ${failedTask.NAME}, error code: ${failedTask.ERROR_CODE}, error message: ${failedTask.ERROR_MESSAGE}`
           )
       })
       expect(records).toEqual([])
   }, 10000)

   afterAll(async () => {
       await pool.destroy()
   })
})

With this approach, we could have the following benefits:

  • Early detection of ETL failures: issues are caught long before they affect downstream
  • Improved data reliability: leads to higher confidence in the data across dashboards and reports
  • Reduced manual effort: engineers do not have to waste time on a manual daily tasks check
  • Proactive incident management: integration with the communication tools such as Slack, Teams, email, etc. allows proactive response and prevents cases in which users notify problems before the engineers
  • Ability to have an overview picture of ETL health over time
  • Improved SLA compliance: Early automated detection helps teams to maintain data availability and freshness commitments in line with business SLAs

End-to-end

In this part, we are covering the whole process of ETL by publishing some test data and verifying that the data is being processed as it is required. We will use an ETL pipeline example from a previous blog and develop our end-to-end test automation example based on that.

In the previous blog, the main source of the data is the Kafka topic. Therefore, our first test step in some of the test environments would be to prepare the test data by publishing it to the Kafka topic. We will do that with Node.js, but of course, feel free to use any other programming language or tool with which you are comfortable.

npm install kafka
npm install jest or npm install jasmine

Next, we need to produce the Data on our ETL data source that is in this example Kafka. So we will use the Kafka producer:

const { Kafka } = require('kafkajs');

class KafkaProducer {

kafka;
producer;

   constructor() {
       this.kafka = new Kafka({ 
          clientId: process.env.YOUR_CLIENT_ID, 
          brokers: process.env.KAFKA_HOST
       });
       this.producer = this.kafka.producer();
   }

   async sendMessage(message) {
       try {
           await this.producer.connect();
           await this.producer.send({
              topic: process.env.YOUR_KAFKA_TOPIC,
              messages: [ { value: message }, ],
       });
       await this.producer.disconnect()
       } catch (error) {
           console.log(error);
       }
   }
}

module.exports = KafkaProducer;

Note: Although this data source issue could be resolved on the ETL side by various services, such as AWS S3 buckets and RabbitMQ, for this blog, we decided to test a data pipeline that uses Kafka as the data source.

We will still need the Snowflake Connector Util class, but with getters for Snowflake queries for staging and the final table. It should look something as the following:

const Snowflake = require('snowflake-promise').Snowflake

const connectionProperties = {
   account: 'TEST_ACCOUNT',
   username: 'TEST_USERNAME',
   role: 'TEST_ROLE',
   database: 'TEST_DB',
   schema: 'TEST_SCHEMA',
   warehouse: 'TEST_WAREHOUSE',
   authenticator: 'SNOWFLAKE_JWT',
   privateKeyPath: process.env.SF_TEST_PRIVATE_KEY,
   privateKeyPass: process.env.SF_TEST_PRIVATE_KEY_PASS,
}

const taskHistoryTable = `${connectionProperties.database}.INFORMATION_SCHEMA.TASK_HISTORY`
const packageStagingTable = `${connectionProperties.database}.PACKAGE_STAGING`
const packageFinalTable =  `${connectionProperties.database}.PACKAGE_FINAL`

class SnowflakeConnectorUtil {
   async snowflakePool() {
       return new Snowflake(connectionProperties)
   }

   async createPool() {
       let pool = await this.snowflakePool()
       await pool.connect()

       return pool
   }

   async getSnowflakeFailedTasksInLast24Hours() {
       return `SELECT *
      FROM (
         SELECT
           distinct name,
           first_value(completed_time) ignore nulls over (partition by name order by completed_time desc) as last_completed_time,
           first_value(state) ignore nulls over (partition by name order by completed_time desc) as last_state
         FROM ${taskHistoryTable}
       )
       WHERE last_state IN ('FAILED', 'CANCELLED', 'FAILED_AND_AUTO_SUSPENDED') AND last_completed_time >= DATEADD(DAY, -1, GETDATE());`
   }

   checkNullValuesQueries = [
       {
          description: 'Package Sender or Recipient ID',
          query: `select * from dbo.schema.packages_final
              where sender_id is null or getter_id is null`,
          expected: 0,
       },
       {
          description: 'User ID',
          query: `select * from dbo.schema.user_request where user_id is null`,
          expected: 0,
       }
   ]

   getPackageStagingTableQuery(packageId) {
       return `select * from ${packageStagingTable} where package_id = '${packageId}'`
   }

   getPackageFinalTableQuery(packageId) {
       return `select * from ${packageFinalTable} where package_id = '${packageId}'`
   }
}

module.exports = SnowflakeConnectorUtil

Then we just need to use everything mentioned in the test case:

import { v6 as uuidv6 } from 'uuid';
const Kafka = require('./KafkaProducer')
const SnowflakeConnectorUtil = require('./db/SnowflakeConnectorUtill')
const kafka = new Kafka()
const snowflakeConnectorUtil = new SnowflakeConnectorUtil()

describe('E2E test example', () => {
   let pool
   let packageId

   beforeAll(async () => {
       pool = await snowflakeConnectorUtil.createPool()
       packageId = uuidv6()
       // producing test data
       await kafka.sendMessage(
       `{
           'packageId', ${packageId},
           'senderId': '1',
           'recipientId': '2',
       }`)
   })

   it ('Verify staging table', async () => {
       const packagesStagingQuery = snowflakeConnectorUtil.getPackageStagingTableQuery(packageId)
       const packagesStaging = await pool.execute(packagesStagingQuery)
       expect(packagesStaging).toBeGreaterThan(0)
       expect(packagesStaging[0].senderFirstName).toBe('John')
       expect(packagesStaging[0].senderLastName).toBe('Doe')
       expect(packagesStaging[0].recipientFirstName).toBe('Jane')
       expect(packagesStaging[0].recipientLastName).toBe('Doe')
   })

   it ('Verify final table', async () => {
       const packagesFinalQuery = snowflakeConnectorUtil.getPackageFinalTableQuery(packageId)
       const packagesFinal = await pool.execute(packagesFinalQuery)
       expect(packagesFinal).toBeGreaterThan(0)
       expect(packagesFinal[0].senderFirstName).toBe('John')
       expect(packagesFinal[0].senderLastName).toBe('Doe')
       expect(packagesFinal[0].recipientFirstName).toBe('Jane')
       expect(packagesFinal[0].recipientLastName).toBe('Doe')
   })

   afterAll(async () => {
       await pool.destroy()
   })
})

By combining various data, we can create multiple test scenarios and achieve confidence in our pipeline after the team makes changes.

This approach leads us to the following benefits:

  • Ensuring data integrity across the pipeline
  • Validates transformations and business logic
  • Detects schema or contract breaks early
  • Builds confidence in data freshness and accuracy
  • Reduces manual verification effort


IQR analysis

You can learn more about the IQR on the following blog IQR in Automation Testing: Unleashing the Data Analytics Potential – Atlantbh Sarajevo. In the context of testing Snowflake ETL pipelines, we can implement this on the appropriate dataset. Test cases based on this approach will provide us more confidence in our ETL process. Significant changes in our data can point to potential issues. A significant increase might indicate that our user base is growing, so we should be prepared for that. A sudden drop could indicate an issue with data processing.

For instance, let us say we want to find outliers in the data set for the final table names, which will be called “package_final”. A common use case would be to identify outliers, such as a significant drop or increase in data over the last 30 days. Since it is not valid in many cases to compare Mondays with Saturdays, we should create an appropriate query to retrieve the last 30 Mondays.

We will add this query to the SnowflakeConnectorUtil class:

getLast30Weeks() {
       return `WITH OrderedRows AS (
           SELECT DATE(created_ts_pst) created_date, count(*) as count, ROW_NUMBER() OVER (ORDER BY created_date) AS row_num, FROM ${packageFinalTable} WHERE
               created_ts_pst BETWEEN DATEADD(DAY, -210, current_date) AND current_date
           GROUP BY created_date
           Order by created_date desc
       )
       SELECT
           *
       FROM
           OrderedRows
       WHERE
           row_num % 7 = 0;`
   }

We will wrap our IQR analyze logic into shared service:

const iqrAnalyze = async data => {
   const dataSetValues = data.map(value => value.COUNT)
   const dataSetSorted = dataSetValues.sort((a, b) => a - b)
   const quartile1 = dataSetSorted[Math.floor(dataSetSorted.length / 4)]
   const quartile3 = dataSetSorted[Math.ceil(dataSetSorted.length * (3 / 4))]
   const iqr = quartile3 - quartile1
   const lowerBound = quartile1 - 1.5 * iqr
   const upperBound = quartile3 + 1.5 * iqr
   const outliers = dataSetSorted.filter(
       value => (value < lowerBound || value > upperBound) && value == dataSet[0].COUNT
   )

   expect(outliers.length).toBe(0)
}

module.exports = {
   iqrAnalyze: iqrAnalyze,
}

Next we test it by using data from Snowflake in our test file:

const SnowflakeConnectorUtil = require('./db/SnowflakeConnectorUtil')
const snowflakeConnectorUtil = new SnowflakeConnectorUtil()
const iqrAnalyzer = require('./services/iqrAnalyze')

describe('IQR test suite', () => {
  let pool

  beforeAll(async () => {
      pool = await snowflakeConnectorUtil.createPool()
  }, 10000)

  it('Packages test', async () => {
   const query = await snowflakeConnectorUtil.getLast30Weeks()
   const data = await pool.execute(query)
   await iqrAnalyzer.iqrAnalyze(data)
  }, 10000)

  afterAll(async () => {
      await pool.destroy()
  })
})

The main advantages of incorporating IQR analysis into automated testing include:

  • Early data anomaly detection
  • Better confidence in data quality
  • Data observability support that leads to better confidence in ETL


Data Quality verification

In this section, we should verify the quality of data that has undergone the pipeline transformations. One of the most common mistakes in ETL processes is invalid data in columns such as NULL values. To find these defects, we could use the following approach.

First, let us define an array of queries, each with a description and the expected number of records that should be returned with the query:

checkNullValuesQueries = [
       {
           description: 'Package Sender or Recipient ID',
           query: `select * from ${packageFinalTable}
               where sender_id is null or getter_id is null`,
           expected: 0,
       },
       {
           description: 'User ID',
           query: `select * from dbo.schema.user_request where user_id is null`,
           expected: 0,
       }
   ]

Next, in the test case we go through the array of queries, execute each one and assert the expected records length:

const SnowflakeConnectorUtil = require('/db/SnowflakeConnectorUtil')
const snowflakeConnectorUtil = new SnowflakeConnectorUtil()

describe('Data quality verification', () => {
  let pool

  beforeAll(async () => {
      pool = await snowflakeConnectorUtil.createPool()
  }, 10000)

  snowflakeConnectorUtil.checkNullValuesQueries.map(checkNullValueQuery => {
       it(`Verify ${checkNullValueQuery.description}`, async () => {
           const nullFieldsRecords = await pool.execute(checkNullValueQuery.query)
           expect(nullFieldsRecords.length)
               .withContext(JSON.stringify(nullFieldsRecords))
               .toBe(checkNullValueQuery.expected)
           await sleep(10000)
       }, 100000)
   })

  afterAll(async () => {
      await pool.destroy()
  })
})

This check strengthens the overall data governance framework by ensuring the standardized quality of the data, guarantees the data completeness, and supports compliance with business and regulatory requests.


CI/CD

For the CI/CD part, we will use Jenkins; therefore, the first step will be setting up the Jenkins server.
The easiest way would be to use Docker and first pull the latest Jenkins Docker image, and run Jenkins on a Docker container by running the following scripts in the terminal:

docker pull jenkins/jenkins:lts

docker run -d --name jenkins-lts -p 8080:8080 -p 50000:50000 -v jenkins_home:/var/jenkins_home jenkins/jenkins:lts

Next, go to your localhost:8080 (or any other port that you have configured) and configure the admin account.

Since we are developing our test cases in Node JS, we need to install the appropriate Jenkins plugin by navigating to Manage Jenkins > Manage Plugins > Available plugins.

Next, find “NodeJS Plugin” and click on the install button. After the plugin installation, we need to configure the Node JS version by going to Manage Jenkins > Tools, finding NodeJS installations, and clicking on Add “NodeJS”. This will open options for choosing the appropriate Node version that you need, as well as additional packages that you want to have on your Jenkins.

Now we will create our Jenkins files in the git repository. The Jenkins file for an end-to-end test case could be such as the following:

pipeline {
   agent any

   tools {
       nodejs 'NodeJS 18'
   }

   triggers {
       cron('H 0 * * *') // Run once a day at midnight
   }

   stages {
       stage('Checkout') {
           steps {
               checkout scm
           }
       }

       stage('Install Dependencies') {
           steps {
               sh 'npm install'
           }
       }

       stage('Run Test') {
           steps {
               sh 'npm test e2e.test.js'
           }
       }
   }
}

Now let us back to the Jenkins dashboard. Click on the “+ New Item” button and then enter the test name for instance: “E2E Snowflake test”, choose “Pipeline” and click the “Ok” button.

Jenkins dashboard

Afterwards, add description, select the “Github project” checkbox, and paste your remote repository link.

In the pipeline section, choose ‘Pipeline script from SCM’ and set your GitHub repository link, along with other details such as credentials, branch name, and script file path.


Then, save the details and back to the main dashboard Jenkins where we should see our job:

Jenkins main dashboard

We can now manually trigger our test or wait for the cron schedule we set up earlier in the script by specifying the appropriate CRON.

Next steps include adding other test cases and setting up Jenkins on a server of your choice. You could read more about Jenkins on our other blogs: Jenkins blogs.

Reporting

For the reporting part, we will introduce the Allure reporting library. First, we need to install the needed dependency in the Node JS repository with the command:

npm i -D jest-allure2-reporter allure-commandline

Then we should define jest.config.js as:

module.exports = {
 reporters: [
   'default',
   [require.resolve('jest-allure2-reporter'), { resultsDir: 'allure-results' }],
 ],
 testEnvironment: require.resolve('jest-allure2-reporter/environment-node'),
 testTimeout: 30000,
};

In package.json modify change scripts config to be such as:

"scripts": {
   "test:ci": "jest --runInBand"
 },

We are using runInBand, which guarantees running multiple test cases as worker processes sequentially (one by one). If we do not use this, the test cases will run in parallel, and we could face race conditions while writing them to the allure reports and have a risk of corrupted reports.

Then, inside your tests, add allure functions for attaching metadata to every test step that you will later need as an explanation in formatting your report. They could be a story, description, step, attachment, etc. Here is an example so you could use as a reference to your other steps:

it ('Verify staging table', async () => {
       allure.story("Staging Table Check");
       allure.description("Verify that staging table data matches expected schema");
       allure.step("Querying and verifying staging table...", async () => {
         const packagesStagingQuery = snowflakeConnectorUtil.getPackageStagingTableQuery(packageId)
         const packagesStaging = await pool.execute(packagesStagingQuery)
         expect(packagesStaging).toBeGreaterThan(0)
         expect(packagesStaging[0].senderFirstName).toBe('John')
         expect(packagesStaging[0].senderLastName).toBe('Doe')
         expect(packagesStaging[0].recipientFirstName).toBe('Jane')
         expect(packagesStaging[0].recipientLastName).toBe('Doe')
       });
       allure.attachment("Staging query result", "SELECT * FROM staging_table", "text/plain");
   })

Note: Pay attention to the “allure.step”, it takes a string and also a callback as arguments, so here we are passing our test step logic as a callback.

After running the test locally, we should get the allure-results folder in our repository with our report file with some additional details.

Let us now include this in our CI/CD part on Jenkins. First, we need to install Allure Jenkins Plugin in Manage Jenkins > Plugins

Allure Jenkins Plugin
Then we need to install Allure Commandline in Manage Jenkins > Tools

Allure Commandline installations
Once that is done, let us modify the Jenkins script in our Git repository so we could add the reporting part at the bottom. It should be such as the following:

pipeline {
    agent any

    tools {
        nodejs 'NodeJS 18'
    }

    triggers {
        cron('H 0 * * *') // Run once a day at midnight
    }

    stages {
        stage('Checkout') {
            steps {
                checkout scm
            }
        }

        stage('Install Dependencies') {
            steps {
                sh 'npm install'
            }
        }

        stage('Run Test') {
            steps {
                sh 'npm run test:ci e2e.test.js'
            }
        }
    }
    post {
    always {
      archiveArtifacts artifacts: 'allure-results/**', allowEmptyArchive: true
    }
    success {
      allure([
        includeProperties: true,
        jdk: '',
        reportBuildPolicy: 'ALWAYS',
        results: [[path: 'allure-results']]
      ])
    }
    unsuccessful {
      allure([
        includeProperties: true,
        jdk: '',
        reportBuildPolicy: 'ALWAYS',
        results: [[path: 'allure-results']]
      ])
    }
  }
}

After running our Jenkins job we should get the following artifacts:

Ent to end test for the Snowflake data pipeline
If you download allure-report.zip you should find the index.html file as allure report.
If you open it in a browser it should look like the following:

allure-report

Of course, you can modify everything by adopting the mentioned allure functions as you need.

Note: Maybe your browser will block opening these metrics because of CORS but you could avoid that by running simple local static server from allure-report folder with python or some other technology by running command:

python3 -m http.server 8081

If you open http://localhost:8081/ you should see the allure report without CORS errors.
Finally, you could add a Jenkins step that sends an allure report as an email to the stakeholders or to make Slack notification. It could be done by shell script or Jenkins plugin “Email Extension.

Conclusion

We have explored various approaches to automated testing for ETL pipelines built in Snowflake, orchestrated through Jenkins CI/CD, and integrated with Allure reporting. All these approaches can help us detect issues early, maintain high data reliability, and continuously improve pipeline performance. Automated testing of the ETL pipeline in Snowflake has transformed our ETL validation from a manual process to a proactive and self-monitoring system.

 

Leave a comment

Your email address will not be published. Required fields are marked *