Data lakes used to have a bad reputation when it comes to data quality. In contrast to data warehouses, data doesn’t need to adhere to any predefined schema before we can load it in. Without proper testing and governance, your data lake can easily turn into a data swamp.

In this article, we’ll look at how to build automated data tests that will be executed any time new data is loaded to a data lake. We’ll also configure SNS-based alerting to get notified about data that deviates from our expectations.

There are so many tools for data profiling and data testing out there. Just to list some of them:

We’ll focus on the open source Python library for validating and profiling data called Great Expectations.

The recommended way of using Great Expectations is to:

If you want to implement all these steps, you can follow the official tutorial. The most important page of the entire documentation lists all the expectations you can use.

If you look at the bullet points above, you may notice that this setup is quite involved. And it doesn’t even cover how to package and deploy the code to make it production-ready, how to set up alerts on failure, how to host and share the data docs, or how to build a repeatable process around it for data testing in data pipelines.

Let’s try to approach it in a more “Pythonic” way. We want to use Great Expectations on data stored in AWS S3. We want data tests to run automatically any time a new file arrives in S3. While Great Expectations provides a data source connector for Athena, it would require running the validation on an entire Athena table rather than validating only a specific batch of data loaded to a data lake. Thus, it would be harder to track down which S3 PUT operation caused anomalies in the respective data source.

As an alternative to the Athena data source, we could configure the expectations suite using pandas and flat files as a data source, but even then, the entire process seems a bit cumbersome.

Let’s look at how we can approach it only using Python — no configuration files. The ultimate goal is to deploy this script to AWS Lambda and configure the function to be triggered on each S3 PUT operation to the desired path in our S3 data lake.

We’ll start by generating an hourly time series with a deliberately chosen range of values. Generating a synthetic dataset will allow us to conveniently insert additional “bad values” and see if our data tests detect those anomalies. Here is an example dataset we will be using:

By and large, any potential anomalies can (and should) be tested. For instance, we can validate:

Here is an example implementation of those tests in a single class. It allows you to run each data test individually as well as run all the tests at once. Additionally, the parse_data_test_result() method sends an SNS email alert on any failed data test.

To create an SNS topic for email alerts, you can run the code below. Then, follow the link from the AWS email to confirm your SNS email subscription.

https://gist.github.com/anna-geller/52a51d2e597d35d76f14429ef2cee6bb

You probably want to test your data locally before moving on to production. The data tests from TimeseriesDataTestRunner can be executed on a local development machine. The code snippet below implements:

When we run this locally, we should receive seven emails similar to this one:

The most efficient way to run those tests automatically is to build a Lambda function with an S3 PUT object event trigger. This way, any time a new file gets uploaded to the specified S3 location, the Lambda function will be automatically triggered to test our data.

To accomplish that, we need a Lambda function that will read the S3 key from the event metadata, read the uploaded file into a pandas DataFrame, and finally run the tests. Here is a simple implementation of that:

To build this Lambda function, we need:

Note that timeseries_data_test_runner.py, timeseries_data_generator.py, and lambda.py are located in the src folder. This is important if you want to use the Dockerfile shown above. The project structure looks as follows (the last file will be explained in the next section):

Now all that is left to do is to:

Creating your Lambda function — image by author

Changing the memory and timeout settings— image by author

Setting the S3 trigger — image by author

Setting the IAM permissions — image by author

Similarly to how we ran data tests locally, we can now start uploading our data to S3 and see the alerts being triggered due to failed data tests on AWS.

If you execute all the tests above, you should receive seven emails from AWS similar to the ones below:

Results of failed data tests being triggered automatically on each upload to S3 — image by author

The biggest advantage of our custom alerts is that they show exactly which S3 file upload caused the specific data test to fail. This is something that seems to be hard to accomplish when using the default implementation of Great Expectations.

Additionally, using purely pandas-based expectations makes testing easier and seems to be more “Pythonic” than working with configuration files and Jupyter notebooks. However, the downside of the presented approach is the lack of data docs. If you care about those, have a look at the official Great Expectations tutorial.

In this article, we looked at various Python libraries for data profiling and testing. We examined two ways of using Great Expectations: the traditional config-file-based method as well as a more “Pythonic” do-it-yourself approach using a custom test runner and email alerts. We then investigated how to execute data tests locally and how to ensure automated test runs after any new file gets uploaded to S3.

Thank you for reading!