etl-patterns

ETL Patterns for Somali Dialect Classifier

Pipeline Architecture

Three-Stage Design

1. Extract (Raw Layer)

• Fetch data from multiple sources
• Store raw, unmodified data
• Maintain source provenance
• Location: data/raw/[source-name]/

2. Transform (Staging/Silver Layer)

• Clean and validate data
• Apply quality filters
• Normalize format
• Location: data/staging/ or data/processed/

3. Load (Gold Layer)

• Prepare for model training
• Split into train/val/test
• Export to final format
• Location: data/final/ or data/gold/

---

Extract Patterns

Source Integration

Pattern 1: Web Scraping (Wikipedia, News)

def extract_from_web(url, source_name):
    """Extract text from web sources"""
    raw_data = fetch_url(url)
    save_raw(raw_data, f'data/raw/{source_name}/')
    return raw_data

Pattern 2: API Integration (HuggingFace, Språkbanken)

def extract_from_api(endpoint, api_key, source_name):
    """Extract from external API"""
    response = requests.get(endpoint, headers={'Authorization': api_key})
    save_raw(response.json(), f'data/raw/{source_name}/')
    return response.json()

Pattern 3: File Upload (Manual Datasets)

def extract_from_file(file_path, source_name):
    """Extract from uploaded files"""
    with open(file_path, 'r', encoding='utf-8') as f:
        raw_data = f.read()
    save_raw(raw_data, f'data/raw/{source_name}/')
    return raw_data

---

Transform Patterns

Pattern 1: Cleaning Pipeline

def transform_text(raw_text):
    """Standard cleaning pipeline"""
    # 1. Remove HTML tags
    text = remove_html_tags(raw_text)

    # 2. Normalize whitespace
    text = ' '.join(text.split())

    # 3. Remove URLs
    text = remove_urls(text)

    # 4. Normalize Unicode
    text = text.encode('utf-8').decode('utf-8')

    return text

Pattern 2: Validation & Filtering

def validate_and_filter(records):
    """Apply quality guardrails"""
    validated = []
    for record in records:
        # Language detection
        if not is_somali(record['text']):
            continue

        # Quality scoring
        score = compute_quality_score(record['text'])
        if score < 5:
            continue

        # Duplicate detection
        if is_duplicate(record['text'], validated):
            continue

        validated.append(record)

    return validated