How To Verify Data Quality On Tables Landed On AWS Data Lake And Data Warehouse.

post image

5 Min Read

The Story

As a data engineer, the responsibility of ensuring data quality is not just a task - it’s a commitment to the integrity of insights derived from the vast volumes of data we manage. The accuracy and dependability of your insights depend on the quality of the data you gather. When working with large datasets in Enterprise Data Lakes (EDL) and Enterprise Data Warehouses (EDW), this is especially true.

Ensuring that the tables that you land on these platforms meet the highest data quality standards is a substantial challenge, but one that can be met with the right tools and techniques.

In this guide/post, I’ll detail how I verify Data Quality on tables landed on AWS EDL and EDW using a simple Python script. The script is designed to automate various critical data validation checks, simplifying the process and empowering our data team to maintain and deliver high-quality data.

Understanding Data Quality

Data quality refers to the overall reliability, accuracy, completeness, and consistency of data within a dataset. It involves ensuring that the data meets predefined standards and is fit for its intended use. High-quality data is free from errors, duplications, and inconsistencies, making it a trustworthy foundation for making informed decisions, conducting analysis, and deriving meaningful insights. The significance of data quality lies in its direct impact on the reliability of business intelligence, analytics, and decision-making processes. Poor data quality can lead to flawed insights, operational inefficiencies, and misguided strategic decisions.

The How

Every morning, I would find myself having to log into the AWS Console and run queries on Athena to verify that the tables were ingested and landed successfully in the Data Lake (EDL) the previous night, thereafter run similar data verification queries on our RedShift cluster via DBeaver to verify if tables were exposed successfully. This mundane task became tedious over time, so I developed a Python script to automate the data validation process for both the EDL and the Data Warehouse (EDW).

The script provides a flexible and comprehensive solution for data validation in the AWS environment. It connects to AWS using either SSO (Single Sign-on) or traditional access key and secret credentials [Not Recommended]. It then dynamically iterates through a list of tables and databases specified in a YAML file, executing SQL queries against the EDL to perform a range of critical data validation checks:

  • Table existence: Verifies that the table exists and verifies that the column names are as expected in the EDL/EDW
  • Count per date: Verifies that the number of rows in the table is consistent with the expected number of rows for the date.
  • Metadata uniqueness: Verifies that the metadata for the table is consistent and unique, ensuring that there are no duplicates.
  • Extraction date validation: Verifies that the extraction date for the table is correct.
  • Field-level consistency: Verifies that the values in the table are valid and consistent.

To ensure data synchronization between the EDL and EDW, my script also establishes a Redshift connection using Azure OAuth for secure authentication and replicates the same suite of validation checks on the EDW. This holistic approach guarantees data integrity and quality across AWS environments.

The Walk-Through

In this section, I will detail the script. First, we need to create a Config.yaml file in the project directory. This file contains AWS services to be used, connection configurations, tables, database and column names.

cat << EOF > Config.yaml
  region: eu-west-1
  sso_profile: $SSO_PROFILE
  credentials_path: $CREDENTIALS_PATH|null
      - name: table_name
        database: glue_database_name
          - id
        max_expected_count: 30000000
      iam: true
      ssl: true
      database: database_name
      host: host
      cluster_identifier: cluster id
      user: user
      credentials_provider: 'BrowserAzureOAuth2CredentialsProvider'
      listen_port: port
      client_id: client_id
      idp_tenant: idp_tenant
      scope: scope_url
      - name: table_name
        database: database
          - id

Once we have the Config.yaml file, we can create the script applying Python’s SOLID principles. The code below details shows an AWS_Credentials class that provides methods for retrieving and reading AWS credentials from a configuration file (~/.aws/credentials).

class AWS_Credentials:
    def __init__(
        self, sso_profile: str, credentials_path: str, region_name: str
    ) -> None:
        self.sso_profile = sso_profile
        self.credentials_path = credentials_path
        self.region_name = region_name

    def config_file_path(self) -> str:
        If the sso_profile is not None, return the sso_profile.
        If the sso_profile is None, check if the credentials_path exists.
        If the credentials_path exists, return the credentials_path.
        If the credentials_path does not exist, raise a FileNotFoundError.
        :return: The path to the config file.

        if self.sso_profile:
            return self.sso_profile
        elif Path(self.credentials_path).exists():
            return Path(self.credentials_path).as_posix()
            raise FileNotFoundError

    def read_credentials(self, **kwargs) -> dict:
        `read_credentials` reads the credentials from a file and returns them as a

            if self.sso_profile:
                return {"sso_profile": self.sso_profile}
                config = configparser.RawConfigParser()
                creds_path = self.config_file_path()
                section = kwargs.get("profile", config.sections()[0])
                return {
                    "aws_access_key_id": config.get(section, "aws_access_key_id"),
                    "aws_secret_access_key": config.get(
                        section, "aws_secret_access_key"
                    "aws_session_token": config.get(section, "aws_session_token"),
        except Exception:

Thereafter, we create another class BotoSession which inherits AWS_Credentials and then provides methods for creating a session, verifying the session and returning a client object for an AWS service.

class BotoSession(AWS_Credentials):
    def __init__(
        sso_profile: str = "",
        credentials_path: str = "",
        region_name: str = "",
        self.sso_profile = sso_profile
        self.credentials_path = credentials_path
        self.region_name = region_name
        super().__init__(sso_profile, credentials_path, region_name)

    def session_client(self, client: str, **kwargs) -> object:
        It reads the credentials from the file, creates a session,
        and returns a client object

        :param client: The AWS service you want to use
        :type client: str

        :return: The session object is being returned.
        creds = self.read_credentials(**kwargs)
        if creds.get("sso_profile"):
            session = boto3.Session(
            session = boto3.Session(
        session = self.verified_session(session, **creds)
        return session.client(client)

    def verified_session(session: boto3.session.Session, **kwargs):
        sts = session.client("sts")
            return session
        except ClientError:
                Your Access Expired After 8 Hours.
                You Have To Update Your AWS Credentials File (typically located in
                    ~/.aws/credentials) With The Correct Credentials.
        except UnauthorizedSSOTokenError:
                ["aws", "sso", "login", "--profile", kwargs.get("sso_profile")]
            return session

Once we have all the basic setup ready, we then create an AthenaClient class that provides methods for executing queries on Amazon Athena and retrieving the query results for validation.

class AthenaClient:
    def __init__(self, boto_session, **kwargs):
        self.athena_client = boto_session.session_client("athena")
        self._s3_output_bucket = self._get_athena_s3_output_bucket(
            boto_session, **kwargs

    def _get_athena_s3_output_bucket(self, boto_session: boto3.Session, **kwargs):
        bucket_name_contains: str = kwargs.get("bucket_name_contains", "athena-output")
        bucket_object_prefix: str = kwargs.get("bucket_object_prefix", "cldedl")

        s3_client = boto_session.session_client("s3")
            buckets = s3_client.list_buckets()
        except ClientError:
            athena_output_bucket = [
                for bucket in buckets["Buckets"]
                if bucket_name_contains in bucket["Name"]

            list_objects = s3_client.list_objects(
                Bucket=athena_output_bucket, Delimiter="/"
            sub_dir = [
                for obj in list_objects["CommonPrefixes"]
                if bucket_object_prefix in obj["Prefix"]
            return Path(athena_output_bucket, sub_dir[0]).as_posix()

    def execute_query(self, query: str, max_executions=10):
        def _get_execution_status(execution_id: str) -> str:
            query_execution_stats = self.athena_client.get_query_execution(
            return query_execution_stats["QueryExecution"]["Status"]["State"]

        if not query:
            print("No query specified")

            query_execution = self.athena_client.start_query_execution(
                ResultConfiguration={"OutputLocation": f"s3://{self._s3_output_bucket}"},
            assert query_execution["ResponseMetadata"]["HTTPStatusCode"] == 200
        except self.athena_client.exceptions.InvalidRequestException as e:
            raise ConnectionError(
                f"Failed to issue: Start query execution command: Reason: {str(e)}"
            while max_executions > 0:
                max_executions -= 1
                exec_status = _get_execution_status(query_execution["QueryExecutionId"])
                if exec_status == "SUCCEEDED":
                elif exec_status in ["FAILED", "CANCELLED"]:
                    raise Exception(
                        "Athena query with the string "
                        f'"{query}" failed or was cancelled'

            if exec_status == "SUCCEEDED":
                    response = self.athena_client.get_query_results(
                except Exception:
                    return response["ResultSet"]["Rows"]
            return False

We also, create a RedShiftClient class that creates a connection to a Redshift database using the provided configuration. Future work, includes converting the RedshiftClient to a Context Manager (for auto-closing the client when done.)

class RedShiftClient:
    def __init__(self, config: dict):
        self.config = config

    def create_connection(self):
        conn: redshift_connector.Connection = redshift_connector.connect(
        conn.autocommit = True
            cursor = conn.cursor()
            result = cursor.execute("SELECT 1")
            assert result.fetchone() == [1], "Could not establish connection!"
            return cursor
        except Exception:
            cursor = None
            del cursor

Then implement additional logic to execute a query using the preferred service, return results and run validation.

class DataValidator:

class ConfigurationLoader:

def process_athena(aws_service: str):

def process_redshift(aws_service: str):

def validate_table(client: object, table_name: str, column: str, database: str, max_expected_count: int):

def arguments():

The Script

Combining all of the classes, we end up with the following main function that loads a configuration file, checks if a specific AWS service is configured, and then processes the service based on the provided arguments.

def main():
    args = arguments()
    config = ConfigurationLoader.load_config(args.config)
    service = args.service

    if config.get("aws_service"):
        aws_service = config["aws_service"]
        if service == "athena" and aws_service.get("athena"):
        elif service == "redshift" and aws_service.get("redshift"):
            print(f"Service '{service}' not found in the configuration.")
        print("AWS service configuration not found in the configuration file.")

if __name__ == "__main__":

Running the Script

Before you run the tool ensure that you can login to AWS via SSO or AWS Credentials.

We do this by setting up SSO to avoid the need to configure credentials each time you need to run the verification process.

Configure SSO using the following:

aws configure sso

After the configuration is complete, export the sso profile name to an envvar.

export SSO_PROFILE=$(cat ~/.aws/config | grep "\[profile " | sed -e 's/\[//g' -e 's/\]//g' -e 's/profile //g')

This will create an SSO profile under ~/.aws/config which will be used for any AWS authentication related processes.

Once, the SSO/Credentials are configured then we can run the script:

python -c Config.yaml --service athena

Interpreting Results

$ python -c Config.yaml --service athena

Attempting to automatically open the SSO authorization page in your default browser.
If the browser does not open or you wish to use a different device to authorize this request, open the following URL:

Then enter the code:

Successfully logged into Start URL: https://<id>
Table: 'table_name': Table exists.
Table: 'table_name': meta_extract_date is recent.
Table: 'table_name': Field-level validation passed.
Table: 'table_name': Metadata duplicates validation passed.
Table: 'table_name': Table count per day validation passed.