Data Type Standardization

1. Profile: detect current types and odd values.
2. Map: define the target types and value mappings.
3. Normalize: clean strings (trim, remove separators, unify case, detect locale).
4. Parse: convert text to typed values (timestamp, decimal, boolean).
5. Cast and enforce: apply target data types and constraints.
6. Validate: check counts, nulls, ranges, failed parses.

-- Goal: date to timestamp (UTC midnight), amount to DECIMAL(12,2), paid to BOOLEAN
WITH raw AS (
  SELECT * FROM orders_raw
), cleaned AS (
  SELECT
    TRIM(order_id) AS order_id,
    -- Normalize date strings then parse
    COALESCE(
      TO_TIMESTAMP(order_date, 'YYYY-MM-DD'),
      TO_TIMESTAMP(order_date, 'MM/DD/YYYY'),
      TO_TIMESTAMP(order_date, 'YYYYMMDD')
    ) AS order_ts_local,
    -- Normalize amount: remove currency symbols and thousand separators
    REGEXP_REPLACE(REGEXP_REPLACE(amount_txt, '[^0-9,.-]', ''), ',', '.') AS amount_norm,
    UPPER(TRIM(paid_status)) AS paid_norm,
    UPPER(NULLIF(TRIM(currency), '')) AS currency_norm
  FROM raw
), standardized AS (
  SELECT
    order_id,
    -- Assume local time is stored already as local midnight; convert to UTC if timezone known
    -- If timezone unknown, document assumption and keep as naive timestamp
    order_ts_local AT TIME ZONE 'UTC' AS order_ts,
    -- Currency normalization: EUR to USD at 1.10 (example static rate)
    CAST(
      CASE 
        WHEN currency_norm = 'EUR' THEN CAST(amount_norm AS DECIMAL(14,4)) * 1.10
        WHEN currency_norm = 'USD' OR currency_norm IS NULL THEN CAST(amount_norm AS DECIMAL(14,4))
        ELSE CAST(amount_norm AS DECIMAL(14,4)) -- fallback, document
      END AS DECIMAL(12,2)
    ) AS amount_usd,
    CASE 
      WHEN paid_norm IN ('Y','YES','TRUE','1') THEN TRUE
      WHEN paid_norm IN ('N','NO','FALSE','0') THEN FALSE
      ELSE NULL
    END AS paid
  FROM cleaned
)
SELECT * FROM standardized;

Note: For real pipelines, apply proper timezone conversion based on source system metadata.

import pandas as pd
import numpy as np

df = pd.DataFrame({
    'user_id': [' 001 ', '002', '003 '],
    'signup': ['2024-03-01', '03/02/2024', '20240303'],
    'opt_in': ['Y', 'false', None],
    'balance': ['1,234.50', '€12,50', 'USD 10.00']
})

# Trim IDs, preserve leading zeros
df['user_id'] = df['user_id'].str.strip()

# Normalize and parse dates to UTC-naive timestamps (document)
df['signup_ts'] = pd.to_datetime(df['signup'], errors='coerce', infer_datetime_format=True)

# Normalize numbers: remove non-numeric except separators, unify decimal symbol
def norm_amount(x):
    if pd.isna(x):
        return np.nan
    s = str(x)
    s = ''.join(ch for ch in s if ch.isdigit() or ch in [',', '.', '-'])
    s = s.replace(',', '.')
    try:
        return float(s)
    except ValueError:
        return np.nan

df['balance_num'] = df['balance'].apply(norm_amount)

# Map booleans
true_set = {'y','yes','true','1','t'}
false_set = {'n','no','false','0','f'}

def to_bool(x):
    if pd.isna(x):
        return np.nan
    v = str(x).strip().lower()
    if v in true_set:
        return True
    if v in false_set:
        return False
    return np.nan

df['opt_in_bool'] = df['opt_in'].apply(to_bool)

# Final dtypes: user_id (string), signup_ts (datetime64[ns]), balance_num (float), opt_in_bool (boolean/nullable)

-- Assume a Spark DataFrame 'df' with columns: dt_str, amount_str, flag
-- Using SQL expressions
SELECT
  TRIM(id) AS id,
  COALESCE(
    to_timestamp(dt_str, 'yyyy-MM-dd HH:mm:ss'),
    to_timestamp(dt_str, 'MM/dd/yyyy HH:mm'),
    to_timestamp(dt_str, 'yyyyMMdd')
  ) AS event_ts,
  CAST(
    regexp_replace(regexp_replace(amount_str, '[^0-9,.-]', ''), ',', '.')
    AS DECIMAL(12,2)
  ) AS amount,
  CASE 
    WHEN lower(flag) IN ('y','yes','true','1') THEN TRUE
    WHEN lower(flag) IN ('n','no','false','0') THEN FALSE
    ELSE NULL
  END AS flag_bool
FROM df;

Source table: raw_payments with columns:

• txn_id STRING (may contain spaces)
• txn_date STRING with mixed formats: '2024-03-01', '03/01/2024', '20240301'
• amount STRING like '1,234.50', '€12,50', 'USD 10.00'
• currency STRING: 'EUR', 'USD', or NULL
• success STRING: 'Y','N','true','false','1','0', or NULL

Task:

• Produce: txn_id (trimmed), txn_ts (TIMESTAMP at UTC midnight), amount_usd DECIMAL(12,2) using rate EUR->USD = 1.10, success BOOLEAN.
• When currency is NULL but amount contains a symbol, infer currency from the symbol.
• Set parsing failures to NULL and keep row counts unchanged.

Expected output: a query returning standardized columns with the specified types.

Input DataFrame raw_users with columns:

• user_id: strings with leading/trailing spaces, some like '007'
• signup_date: mixed forms '01-03-2024', '2024/03/02', 1709337600 (epoch seconds)
• email_opt_in: 'Yes','no','1','0', None
• balance: '1 234,00', '1234.00', ''

Task:

• user_id trimmed (keep leading zeros).
• signup_ts as pandas datetime in UTC; treat epoch as seconds.
• email_opt_in_bool as boolean with unmapped values to NaN.
• balance_num as float with correct decimal point; blanks to NaN.

Expected output: DataFrame with dtypes string, datetime64[ns, UTC], boolean/nullable, and float.

• Size: map variants to S/M/L; blanks to NULL.
• Weight: strip spaces, lowercase, parse unit (g/kg), convert to kg, unify decimal.

-- Pseudocode / SQL-ish
SELECT
  CASE
    WHEN LOWER(TRIM(product_size)) IN ('s','small') THEN 'S'
    WHEN LOWER(TRIM(product_size)) IN ('m','med','medium') THEN 'M'
    WHEN LOWER(TRIM(product_size)) IN ('l','large') THEN 'L'
    ELSE NULL
  END AS size,
  CAST(
    CASE WHEN LOWER(weight) LIKE '%kg%'
      THEN CAST(REPLACE(LOWER(REGEXP_REPLACE(weight, '[^0-9,.-kg ]', '')), ',', '.') AS DOUBLE)
           -- remove 'kg' afterward or extract number first depending on engine
      WHEN LOWER(weight) LIKE '%g%'
      THEN (CAST(REPLACE(LOWER(REGEXP_REPLACE(weight, '[^0-9,.-g ]', '')), ',', '.') AS DOUBLE) / 1000.0)
      ELSE NULL
    END AS DECIMAL(8,3)
  ) AS weight_kg
FROM src;