Data quality management (DQM) is a set of practices aimed at improving and maintaining the quality of an organisation's data. Effective DQM practices help ensure data accuracy, completeness, consistency, timeliness, uniqueness, and validity.
In this article, I explore how to leverage Python libraries to carry out two fundamental DQM procedures: data validation and data cleansing.
The dataset chosen for this exercise is the Kaggle dataset: Cafe Sales - Dirty Data for Cleaning Training by Ahmed Mohamed, under the CC BY-SA 4.0 license. It is a CSV file that comprises 10,000 cafe sales transactions across eight columns. Data quality issues are deliberately introduced, making it well suited for practicing DQM procedures.
To implement these procedures, Pandas is utilized for data cleansing, while Pandera is used for data validation.
Data validation requires establishing and enforcing business rules and checks, according to IBM's data validation page. Each organization defines its own rules and checks. Nevertheless, we'll focus on the most common checks:
- Data type checks
- Uniqueness checks
- Format checks
- Code checks
- Consistency checks
- Range checks
First, I created a data directory inside the Python application to store and access the CSV file. Next, I separated the extraction stage from the transformation stage by creating a dedicated module for each. Visit the GitHub repository to see the project structure.
The execution is applied in the main.py file:
from workflow.extract import extract_data
from workflow.transform import transform_data
def main():
df = extract_data("data/dirty_cafe_sales.csv")
transform_data(df)
main()
Extract the CSV file: extract.py
import pandas as pd
def extract_data(file_path):
df = pd.read_csv(file_path)
return df
DQM procedures - data validation and data cleansing: transform.py
I begin by creating a transform_data(df) function that accepts the DataFrame returned by extract_data in extract.py as a parameter.
Rename columns
df.rename(columns={'Transaction ID': 'transaction_id', 'Item': 'item', 'Quantity': 'quantity', 'Price Per Unit': 'price_per_unit', 'Total Spent': 'total_spent', 'Payment Method': 'payment_method', 'Location': 'location', 'Transaction Date': 'transaction_date'}, inplace=True)
Drop duplicates
df = df.drop_duplicates()
This removes the rows that are 100% identical across all columns.
Extract dirty data
def data_extraction():
dirty_values = ["UNKNOWN", "ERROR"]
dirty_rows = []
rows_to_drop = []
for index, row in df.iterrows():
for value in row:
if value in dirty_values or pd.isna(value):
dirty_rows.append(row)
rows_to_drop.append(index)
break
dirty_dataframe = pd.DataFrame(dirty_rows)
clean_dataframe = df.drop(index=rows_to_drop)
clean_dataframe = clean_dataframe.reset_index(drop=True)
return clean_dataframe, dirty_dataframe
cd, dd = data_extraction()
The objective of data_extraction() is to separate clean rows from those containing data quality issues.
The function first defines dirty_values, which is a list of predefined invalid values (UNKNOWN, ERROR). It then iterates through each row of the DataFrame examining each cell.
If a cell value matches the predefined ones in dirty_values or is identified as a missing value (NaN), it is therefore classified as a "dirty row".
The dirty row is appended to the dirty_rows list, while its index in the original DataFrame df is stored in the rows_to_drop list.
Once all rows are processed, dirty_rows is then converted to the DataFrame dirty_dataframe. Figure 1 presents this DataFrame with all its records.
Figure 1. Visualization of dirty_dataframe and all its records containing data quality issues. This DataFrame stores 6911 rows extracted from the original DataFrame.
These dirty rows whose indexes are stored in rows_to_drop are removed from df, resulting in a new Dataframe: clean_dataframe. Finally, the index of clean_dataframe is reset to maintain sequential row numbering. Figure 2 presents the clean_dataframe.
Figure 2. Visualization of clean_dataframe and all its records containing no data quality issues. This DataFrame stores 3089 rows remaining from the original DataFrame.
In the next steps, we will focus only on clean_dataframe.
Data type conversion
The relevant columns are converted to their appropriate data types. This ensures that numeric values are represented as integers or floats, while dates are stored using Pandas' datetime type.
The data type conversion is implemented as follows:
cd["quantity"] = cd["quantity"].astype(int)
cd["price_per_unit"] = cd["price_per_unit"].astype(float)
cd["total_spent"] = cd["total_spent"].astype(float)
cd["transaction_date"] = pd.to_datetime(cd["transaction_date"], errors="coerce")
Data Type Check
To begin with, however, what is a data type check?
A data type check identifies values that violate the specified data type or do not conform to its expected length, precision, or scale. Therefore, after performing the data type conversion, Pandera is used to validate that the columns have the expected data types.
Pandera provides two approaches to data validation: DataFrame Model and DataFrame Schema.
For this exercise, I chose to use DataFrame Schema:
data_type_check = pa.DataFrameSchema(
{
"transaction_id": pa.Column(str),
"item": pa.Column(str),
"quantity": pa.Column(int),
"price_per_unit": pa.Column(float),
"total_spent": pa.Column(float),
"payment_method": pa.Column(str),
"location": pa.Column(str),
"transaction_date": pa.Column(DateTime)
}
)
data_type_check.validate(cd)
After converting each relevant column to its appropriate data type, the data_type_check schema is used to validate that the DataFrame complies with its rules.
Note: Data validation can be performed either before or after data transformation, depending on the ETL pipeline design. In this case, the schema is applied after the type conversion to verify that the transformation was successful. Alternatively, it can be used before the data type conversion to identify records whose data types do not conform to the expected rules.
Uniqueness Check
A uniqueness check is applied to columns whose values must be unique.
In our case, the uniqueness check can logically be applied only to the transaction_id column. The other columns can store duplicate values.
uniqueness_check = pa.DataFrameSchema(
{
"transaction_id": pa.Column(unique=True),
}
)
uniqueness_check.validate(cd)
Format Check
A format check is applied to columns that require specific data formatting such as email addresses or phone numbers.
In our case, we conduct a format check on the transaction_id column to verify that all the values start with TXN_.
format_check = pa.DataFrameSchema(
{
"transaction_id": pa.Column(
str,
checks=pa.Check(lambda s: s.str.startswith("TXN_"))
)
}
)
format_check.validate(cd)
Code Check
A code check determines whether a data value is valid by comparing it to a list of acceptable values.
In our case, we want to validate the payment_method column against a set of acceptable codes.
transaction_methods = [
"Credit Card",
"Digital Wallet",
"Cash",
"Other",
"Cheque",
"Bank Transfers"
]
code_check = pa.DataFrameSchema(
{
"payment_method": pa.Column(
str,
checks=pa.Check.isin(transaction_methods)
)
}
)
code_check.validate(cd)
We create the transaction_methods list, which contains a set of valid payment methods. Then, we define the DataFrameSchema code_check that uses a validation check to verify that a value is in transaction_methods.
Consistency Check
Consistency checks are performed to verify that the relationships between two or more columns satisfy business rules.
In this exercise, we examine whether the total_spent values are consistent with the result of multiplying quantity by price_per_unit for each row.
consistency_check = pa.DataFrameSchema(
checks=[
pa.Check(lambda df: (
df["total_spent"] == df["quantity"] * df["price_per_unit"]
)
)
]
)
consistency_check.validate(cd)
This schema uses a lambda function to define a custom validation rule.
Range Check
A range check is a data validation check that determines whether numerical data falls within a predefined range of minimum and maximum values.
range_check = pa.DataFrameSchema(
{
"transaction_date": pa.Column(
DateTime,
checks=pa.Check(
lambda s: (
(s >= pd.Timestamp("2023-01-01")) &
(s < pd.Timestamp("2024-01-01"))
)
)
),
"quantity": pa.Column(
int,
checks=pa.Check.ge(0)
),
"price_per_unit": pa.Column(
float,
checks=pa.Check.ge(0)
),
"total_spent": pa.Column(
float,
checks=pa.Check.ge(0)
)
}
)
range_check.validate(cd)
We examine whether quantity, price_per_unit, and total_spent contain only non-negative values. We also verify that the dataset only includes rows with dates between January 1st, 2023, and December 31st, 2023.
Sort the DataFrame by transaction date in descending order
cd_sorted = cd.sort_values(by='transaction_date', ascending=False)
Figure 3 presents the result of this transformation.
Figure 3. Visualization of cd_sorted after conducting a sort by transaction_date in descending order. This DataFrame stores 3089 rows remaining from the original DataFrame.
Results
The results of these DQM procedures are as follows:
-
cd_sorted: a clean DataFrame containing no identified data quality issues. -
dd: a dirty DataFrame containing the rows with data quality issues extracted from the original dataset. The objective was not to permanently discard these records. Instead, they are preserved inddas they may contain valuable information that can be used for further investigation.
The two distinct DataFrames can then be loaded into the destination of choice.
References
Top comments (0)