Dagster + Great Expectations: The Open Source Duo Transforming Data Quality Pipelines

Ensuring data quality is paramount for businesses relying on data-driven decision-making. As data volumes grow and sources diversify, manual quality checks become increasingly impractical and error-prone. This is where automated data quality checks come into play, offering a scalable solution to maintain data integrity and reliability. At my organization, which collects large volumes of public web data, we’ve developed a robust system for automated data quality checks using two powerful open-source tools: Dagster and Great Expectations. These tools are the cornerstone of our approach to data quality management, allowing us to efficiently validate and monitor our data pipelines at scale. Dagster Dagster Great Expectations Great Expectations In this article, I’ll explain how we use Dagster, an open-source data orchestrator, and Great Expectations, a data validation framework, to implement comprehensive automated data quality checks. I’ll also explore the benefits of this approach and provide practical insights into our implementation process, including a Gitlab demo, to help you understand how these tools can enhance your own data quality assurance practices. Let’s discuss each of them in more detail before moving to practical examples. Dagster: An Open-Source Data Orchestrator Used for ETL, analytics, and machine learning workflows, Dagster lets you build, schedule, and monitor data pipelines. This Python-based tool allows data scientists and engineers to easily debug runs, inspect assets, or get details about their status, metadata, or dependencies. As a result, Dagster makes your data pipelines more reliable, scalable, and maintainable. It can be deployed in Azure, Google Cloud, AWS, and many other tools you may already be using. Airflow and Prefect can be named as Dagster competitors, but I personally see more pros in the latter, and you can find plenty of comparisons online before committing. Great Expectations: A Data Validation Framework A great tool with a great name, Great Expectations is an open-source platform for maintaining data quality. This Python library actually uses “Expectation” as their in-house term for assertions about data. Great Expectations provides validations based on the schema and values. Some examples of such rules could be max or min values and count validations. It also provides data validation and can generate expectations according to the input data. Of course, this feature usually requires some tweaking, but it definitely saves some time. Another useful aspect is that Great Expectations can be integrated with Google Cloud, Snowflake, Azure, and over 20 other tools. While it can be challenging for data users without technical knowledge, it’s nevertheless worth attempting. Why are Automated Data Quality Checks Necessary? Automated quality checks have multiple benefits for businesses that handle voluminous data of critical importance. If the information must be accurate, complete, and consistent, automation will always beat manual labor, which is prone to errors. Let’s take a quick look at the 5 main reasons why your organization might need automated data quality checks. 1. Data integrity Your organization can collect reliable data with a set of predefined quality criteria. This reduces the chance of wrong assumptions and decisions that are error-prone and not data-driven. Tools like Great Expectations and Dagster can be very helpful here. 2. Error minimization While there’s no way to eradicate the possibility of errors, you can minimize the chance of them occurring with automated data quality checks. Most importantly, this will help identify anomalies earlier in the pipeline, saving precious resources. In other words, error minimization prevents tactical mistakes from becoming strategic. 3. Efficiency Checking data manually is often time-consuming and may require more than one employee on the job. With automation, your data team can focus on more important tasks, such as finding insights and preparing reports. 4. Real-time monitoring Automatization comes with a feature of real-time tracking. This way, you can detect issues before they become bigger problems. In contrast, manual checking takes longer and will never catch the error at the earliest possible stage. 5. Compliance Most companies that deal with public web data know about privacy-related regulations. In the same way, there may be a need for data quality compliance, especially if it later goes on to be used in critical infrastructure, such as pharmaceuticals or the military. When you have automated data quality checks implemented, you can give specific evidence about the quality of your information, and the client has to check only the data quality rules but not the data itself. How We Test Data Quality As a public web data provider, having a well-oiled automated data quality check mechanism is key. So how do we do it? First, we differentiate our tests by the type of data. The test naming might seem somewhat confusing because it was originally conceived for internal use, but it helps us to understand what we’re testing. We have two types of data: Static data. Static means that we don’t scrape the data in real-time but rather use a static fixture. Dynamic data. Dynamic means that we scrape the data from the web in real-time. Static data. Static means that we don’t scrape the data in real-time but rather use a static fixture. Static data. Dynamic data. Dynamic means that we scrape the data from the web in real-time. Dynamic data. Then, we further differentiate our tests by the type of data quality check: Fixture tests. These tests use fixtures to check the data quality. Coverage tests. These tests use a bunch of rules to check the data quality. Fixture tests. These tests use fixtures to check the data quality. Fixture tests. Coverage tests. These tests use a bunch of rules to check the data quality. Coverage tests. Let’s take a look at each of these tests in more detail. Static fixture tests As mentioned earlier, these tests belong to the static data category, meaning we don’t scrape the data in real-time. Instead, we use a static fixture that we have saved previously. A static fixture is input data that we have saved previously. In most cases, it’s an HTML file of a web page that we want to scrape. For every static fixture, we have a corresponding expected output. This expected output is the data that we expect to get from the parser. The test works like this: The parser receives the static fixture as an input. The parser processes the fixture and returns the output. The test checks if the output is the same as the expected output. This is not a simple JSON comparison because some fields are expected to change (such as the last updated date), but it is still a simple process. The parser receives the static fixture as an input. The parser processes the fixture and returns the output. The test checks if the output is the same as the expected output. This is not a simple JSON comparison because some fields are expected to change (such as the last updated date), but it is still a simple process. We run this test in our CI/CD pipeline on merge requests to check if the changes we made to the parser are valid and if the parser works as expected. If the test fails, we know we have broken something and need to fix it. Static fixture tests are the most basic tests both in terms of process complexity and implementation because they only need to run the parser with a static fixture and compare the output with the expected output using a rather simple Python script. Nevertheless, they are still really important because they are the first line of defense against breaking changes. However, a static fixture test cannot check whether scraping is working as expected or whether the page layout remains the same. This is where the dynamic tests category comes in. Dynamic fixture tests Basically, dynamic fixture tests are the same as static fixture tests, but instead of using a static fixture as an input, we scrape the data in real-time. This way, we check not only the parser but also the scraper and the layout. Dynamic fixture tests are more complex than static fixture tests because they need to scrape the data in real-time and then run the parser with the scraped data. This means that we need to launch both the scraper and the parser in the test run and manage the data flow between them. This is where Dagster comes in. Dagster is an orchestrator that helps us to manage the data flow between the scraper and the parser. There are four main steps in the process: Seed the queue with the URLs we want to scrape. Scrape. Parse. Check the parsed document against the saved fixture. Seed the queue with the URLs we want to scrape. Scrape. Parse. Check the parsed document against the saved fixture. The last step is the same as in static fixture tests, and the only difference is that instead of using a static fixture, we scrape the data during the test run. Dynamic fixture tests play a very important role in our data quality assurance process because they check both the scraper and the parser. Also, they help us understand if the page layout has changed, which is impossible with static fixture tests. This is why we run dynamic fixture tests in a scheduled manner instead of running them on every merge request in the CI/CD pipeline. However, dynamic fixture tests do have a pretty big limitation. They can only check the data quality of the profiles over which we have control. For example, if we don’t control the profile we use in the test, we can’t know what data to expect because it can change anytime. This means that dynamic fixture tests can only check the data quality for websites in which we have a profile. To overcome this limitation, we have dynamic coverage tests. Dynamic coverage tests Dynamic coverage tests also belong to the dynamic data category, but they differ from dynamic fixture tests in terms of what they check. While dynamic fixture tests check the data quality of the profiles we have control over, which is pretty limited because it is not possible in all targets, dynamic coverage tests can check the data quality without a need to control the profile. This is possible because dynamic coverage tests don’t check the exact values, but they check those against a set of rules we have defined. This is where Great Expectations comes in. Dynamic coverage tests are the most complex tests in our data quality assurance process. Dagster also orchestrates them as dynamic fixture tests. However, we use Great Expectations instead of a simple Python script to execute the test here. At first, we need to select the profiles we want to test. Usually, we select profiles from our database that have high field coverage. We do this because we want to ensure the test covers as many fields as possible. Then, we use Great Expectations to generate the rules using the selected profiles. These rules are basically the constraints that we want to check against the data. Here are some examples: All profiles must have a name. At least 50% of the profiles must have a last name. The education count value cannot be lower than 0. All profiles must have a name. At least 50% of the profiles must have a last name. The education count value cannot be lower than 0. After we have generated the rules, called expectations in Great Expectations, we can run the test pipeline, which consists of the following steps: Seed the queue with the URLs we want to scrape. Scrape. Parse. Validate parsed documents using Great Expectations. Seed the queue with the URLs we want to scrape. Scrape. Parse. Validate parsed documents using Great Expectations. This way, we can check the data quality of profiles over which we have no control. Dynamic coverage tests are the most important tests in our data quality assurance process because they check the whole pipeline from scraping to parsing and validate the data quality of profiles over which we have no control. This is why we run dynamic coverage tests in a scheduled manner for every target we have. However, implementing dynamic coverage tests from scratch can be challenging because it requires some knowledge about Great Expectations and Dagster. This is why we have prepared a demo project showing how to use Great Expectations and Dagster to implement automated data quality checks. Implementing Automated Data Quality Checks In this Gitlab repository, you can find a demo of how to use Dagster and Great Expectations to test data quality. The dynamic coverage test graph has more steps, such as seed_urls, scrape, parse, and so on, but for the sake of simplicity, in this demo, some operations are omitted. However, it contains the most important part of the dynamic coverage test — data quality validation. The demo graph consists of the following operations: In this Gitlab repository In this Gitlab repository seed_urls scrape parse load_items: loads the data from the file and loads them as JSON objects. load_structure : loads the data structure from the file. get_flat_items : flattens the data. load_dfs : loads the data as Spark DataFrames by using the structure from the load_structure operation. ge_validation : executes the Great Expectations validation for every DataFrame. post_ge_validation : checks if the Great Expectations validation passed or failed. load_items: loads the data from the file and loads them as JSON objects. load_items load_structure : loads the data structure from the file. load_structure get_flat_items : flattens the data. get_flat_items load_dfs : loads the data as Spark DataFrames by using the structure from the load_structure operation. load_dfs ge_validation : executes the Great Expectations validation for every DataFrame. ge_validation post_ge_validation : checks if the Great Expectations validation passed or failed. post_ge_validation While some of the operations are self-explanatory, let’s examine some that might require further detail. Generating a structure The load_structure operation itself is not complicated. However, what is important is the type of structure. It’s represented as a Spark schema because we will use it to load the data as Spark DataFrames because Great Expectations works with them. Every nested object in the Pydantic model will be represented as an individual Spark schema because Great Expectations doesn’t work well with nested data. load_structure For example, a Pydantic model like this: python class CompanyHeadquarters(BaseModel): city: str country: str class Company(BaseModel): name: str headquarters: CompanyHeadquarters python class CompanyHeadquarters(BaseModel): city: str country: str class Company(BaseModel): name: str headquarters: CompanyHeadquarters This would be represented as two Spark schemas: json { "company": { "fields": [ { "metadata": {}, "name": "name", "nullable": false, "type": "string" } ], "type": "struct" }, "company_headquarters": { "fields": [ { "metadata": {}, "name": "city", "nullable": false, "type": "string" }, { "metadata": {}, "name": "country", "nullable": false, "type": "string" } ], "type": "struct" } } json { "company": { "fields": [ { "metadata": {}, "name": "name", "nullable": false, "type": "string" } ], "type": "struct" }, "company_headquarters": { "fields": [ { "metadata": {}, "name": "city", "nullable": false, "type": "string" }, { "metadata": {}, "name": "country", "nullable": false, "type": "string" } ], "type": "struct" } } The demo already contains data, structure, and expectations for Owler company data. However, if you want to generate a structure for your own data (and your own structure), you can do that by following the steps below. Run the following command to generate an example of the Spark structure: docker run -it - rm -v $(pwd)/gx_demo:/gx_demo gx_demo /bin/bash -c "gx structure" docker run -it - rm -v $(pwd)/gx_demo:/gx_demo gx_demo /bin/bash -c "gx structure" This command generates the Spark structure for the Pydantic model and saves it as example_spark_structure.json in the gx_demo/data directory. example_spark_structure.json gx_demo/data Preparing and validating data After we have the structure loaded, we need to prepare the data for validation. That leads us to the get_flat_items operation, which is responsible for flattening the data. We need to flatten the data because each nested object will be represented as a row in a separate Spark DataFrame. So, if we have a list of companies that looks like this: get_flat_items json [ { "name": "Company 1", "headquarters": { "city": "City 1", "country": "Country 1" } }, { "name": "Company 2", "headquarters": { "city": "City 2", "country": "Country 2" } } ] json [ { "name": "Company 1", "headquarters": { "city": "City 1", "country": "Country 1" } }, { "name": "Company 2", "headquarters": { "city": "City 2", "country": "Country 2" } } ] After flattening, the data will look like this: json { "company": [ { "name": "Company 1" }, { "name": "Company 2" } ], "company_headquarters": [ { "city": "City 1", "country": "Country 1" }, { "city": "City 2", "country": "Country 2" } ] } json { "company": [ { "name": "Company 1" }, { "name": "Company 2" } ], "company_headquarters": [ { "city": "City 1", "country": "Country 1" }, { "city": "City 2", "country": "Country 2" } ] } Then, the flattened data from the get_flat_items operation will be loaded into each Spark DataFrame based on the structure that we loaded in the load_structure operation in the load_dfs operation. get_flat_items load_structure load_dfs The load_dfs operation uses DynamicOut, which allows us to create a dynamic graph based on the structure that we loaded in the load_structure operation. load_dfs DynamicOut DynamicOut load_structure Basically, we will create a separate Spark DataFrame for every nested object in the structure. Dagster will create a separate ge_validation operation that parallelizes the Great Expectations validation for every DataFrame. Parallelization is useful not only because it speeds up the process but also because it creates a graph to support any kind of data structure. ge_validation So, if we scrape a new target, we can easily add a new structure, and the graph will be able to handle it. Generate expectations Expectations are also already generated in the demo and the structure. However, this section will show you how to generate the structure and expectations for your own data. Make sure to delete previously generated expectations if you’re generating new ones with the same name. To generate expectations for the gx_demo/data/owler_company.json data, run the following command using gx_demo Docker image: gx_demo/data/owler_company.json gx_demo docker run -it - rm -v $(pwd)/gx_demo:/gx_demo gx_demo /bin/bash -c "gx expectations /gx_demo/data/owler_company_spark_structure.json /gx_demo/data/owler_company.json owler company" docker run -it - rm -v $(pwd)/gx_demo:/gx_demo gx_demo /bin/bash -c "gx expectations /gx_demo/data/owler_company_spark_structure.json /gx_demo/data/owler_company.json owler company" The command above generates expectations for the data (gx_demo/data/owler_company.json) based on the flattened data structure (gx_demo/data/owler_company_spark_structure.json). In this case, we have 1,000 records of Owler company data. It’s structured as a list of objects, where each object represents a company. gx_demo/data/owler_company.json gx_demo/data/owler_company_spark_structure.json After running the above command, the expectation suites will be generated in the gx_demo/great_expectations/expectations/owler directory. There will be as many expectation suites as the number of nested objects in the data, in this case, 13. gx_demo/great_expectations/expectations/owler Each suite will contain expectations for the data in the corresponding nested object. The expectations are generated based on the structure of the data and the data itself. Keep in mind that after Great Expectations generates the expectation suite, which contains expectations for the data, some manual work might be needed to tweak or improve some of the expectations. Let’s take a look at the 6 generated expectations for thefollowers field in the company suite: followers company expect_column_min_to_be_between expect_column_max_to_be_between expect_column_mean_to_be_between expect_column_median_to_be_between expect_column_values_to_not_be_null expect_column_values_to_be_in_type_list expect_column_min_to_be_between expect_column_min_to_be_between expect_column_max_to_be_between expect_column_max_to_be_between expect_column_mean_to_be_between expect_column_mean_to_be_between expect_column_median_to_be_between expect_column_median_to_be_between expect_column_values_to_not_be_null expect_column_values_to_not_be_null expect_column_values_to_be_in_type_list expect_column_values_to_be_in_type_list We know that the followers field represents the number of followers of the company. Knowing that, we can say that this field can change over time, so we can’t expect the maximum value, mean, or median to be the same. followers However, we can expect the minimum value to be greater than 0 and the values to be integers. We can also expect that the values are not null because if there are no followers, the value should be 0. So, we need to get rid of the expectations that are not suitable for this field: expect_column_max_to_be_between, expect_column_mean_to_be_between, and expect_column_median_to_be_between. expect_column_max_to_be_between expect_column_mean_to_be_between expect_column_median_to_be_between However, every field is different, and the expectations might need to be adjusted accordingly. For example, the field completeness_score represents the company’s completeness score. For this field, it makes sense to expect the values to be between 0 and 100, so we can keep not only expect_column_min_to_be_between but also expect_column_max_to_be_between. completeness_score expect_column_min_to_be_between expect_column_max_to_be_between Take a look at the Gallery of Expectations to see what kind of expectations you can use for your data. Gallery of Expectations Gallery of Expectations Running the demo To see everything in action, go to the root of the project and run the following commands: Build Docker image: Build Docker image: docker build -t gx_demo docker build -t gx_demo Run Docker container: Run Docker container: docker composer up docker composer up After running the above commands, the Dagit (Dagster UI) will be available at localhost:3000. Run the demo_coverage job with the default configuration from the launchpad. After the job execution, you should see dynamically generated ge_validation operations for every nested object. localhost:3000 demo_coverage ge_validation In this case, the data passed all the checks, and everything is beautiful and green. If data validation for any nested object fails, then postprocess_ge_validation operations would be marked as failed (and obviously, it would be red instead of green). Let’s say the company_ceo validation failed. The postprocess_ge_validation[company_ceo] operation would be marked as failed. To see what expectations failed particularly, click on the ge_validation[company_ceo] operation and open “Expectation Results” by clicking on the “[Show Markdown]” link. It will open the validation results overview modal with all the data about the company_ceo dataset. postprocess_ge_validation company_ceo postprocess_ge_validation[company_ceo] ge_validation[company_ceo] company_ceo Conclusion Depending on the stage of the data pipeline, there are many ways to test data quality. However, it is essential to have a well-oiled automated data quality check mechanism to ensure the accuracy and reliability of the data. Tools like Great Expectations and Dagster aren’t strictly necessary (static fixture tests don’t use any of those), but they can greatly help with a more robust data quality assurance process. Whether you’re looking to enhance your existing data quality processes or build a new system from scratch, we hope this guide has provided valuable insights.