Common Data Quality Issues — and How Data Engineers Fix Them

import pandas as pd

df = pd.read_csv('customers.csv')

# Full NULL analysis across all columns
null_report = pd.DataFrame({
    'total_rows': len(df),
    'null_count': df.isnull().sum(),
    'null_pct': (df.isnull().sum() / len(df) * 100).round(2),
    'dtype': df.dtypes,
}).sort_values('null_pct', ascending=False)

print(null_report)
# Shows which columns have most NULLs — prioritize fixing those

# Raise an alert if critical columns exceed threshold
CRITICAL_COLUMNS = ['user_id', 'email', 'created_at']
NULL_THRESHOLD_PCT = 1.0  # >1% NULLs in critical columns is a pipeline failure

for col in CRITICAL_COLUMNS:
    null_pct = df[col].isnull().mean() * 100
    if null_pct > NULL_THRESHOLD_PCT:
        raise ValueError(
            f"Data quality failure: {col} has {null_pct:.1f}% NULLs "
            f"(threshold: {NULL_THRESHOLD_PCT}%)"
        )

import pandas as pd

df = pd.read_csv('customers.csv')

# Detect duplicates by different keys
print(f"Total rows: {len(df)}")
print(f"Exact duplicate rows: {df.duplicated().sum()}")
print(f"Email duplicates (business key): {df.duplicated(subset=['email']).sum()}")
print(f"Phone duplicates: {df.duplicated(subset=['phone']).sum()}")

# Examine duplicate examples
email_dupes = df[df.duplicated(subset=['email'], keep=False)]
print("\nSample email duplicates:")
print(email_dupes.sort_values('email').head(10))

# Strategy 1: Keep most recent record per email
df_deduped = (
    df.sort_values('updated_at', ascending=False)
    .drop_duplicates(subset=['email'], keep='first')
    .reset_index(drop=True)
)
print(f"\nAfter dedup: {len(df_deduped)} rows ({len(df) - len(df_deduped)} removed)")

# Strategy 2: Merge duplicate records (consolidate data from all copies)
def merge_duplicates(group):
    """Combine duplicate records: take most recent non-null value per field"""
    result = {}
    for col in group.columns:
        # Take first non-null value from the group (sorted by recency)
        non_null = group[col].dropna()
        result[col] = non_null.iloc[0] if len(non_null) > 0 else None
    return pd.Series(result)

df_merged = (
    df.sort_values('updated_at', ascending=False)
    .groupby('email', as_index=False)
    .apply(merge_duplicates)
)

# Strategy 3: Flag for manual review (don't auto-delete)
df['is_duplicate'] = df.duplicated(subset=['email'], keep='first')
duplicates_for_review = df[df['is_duplicate']].to_csv('duplicates_review.csv')

import great_expectations as gx

context = gx.get_context()

# Define comprehensive expectations for your dataset
suite = context.add_expectation_suite("orders_suite")

validator = context.get_validator(
    batch_request=batch_request,
    expectation_suite_name="orders_suite"
)

# 1. Column existence (schema drift detection)
for col in ["order_id", "customer_id", "amount", "status", "created_at"]:
    validator.expect_column_to_exist(col)

# 2. Not-null constraints on critical fields
validator.expect_column_values_to_not_be_null("order_id")
validator.expect_column_values_to_not_be_null("amount")

# 3. Value range and type constraints
validator.expect_column_values_to_be_between("amount", min_value=0, max_value=1_000_000)
validator.expect_column_values_to_be_in_set(
    "status", {"pending", "processing", "shipped", "delivered", "cancelled"}
)
validator.expect_column_values_to_match_regex("order_id", r"^ORD-\d{8}$")

# 4. Uniqueness
validator.expect_column_values_to_be_unique("order_id")

# 5. Row count sanity check (catches truncated loads)
validator.expect_table_row_count_to_be_between(min_value=1000, max_value=10_000_000)

# 6. Freshness check (data must be recent)
validator.expect_column_max_to_be_between(
    "created_at",
    min_value="2026-01-01",  # must have recent data
)

# Run validation — pipeline fails if expectations not met
results = validator.validate()
if not results["success"]:
    failed = [r for r in results.results if not r.success]
    raise ValueError(f"Data quality failures:\n" +
                     "\n".join(f"  - {r.expectation_config.expectation_type}: {r.result}"
                               for r in failed))