Building the Bronze Layer: Scalable Data Ingestion

flowchart TD
    A[Source: MTSamples] -->|Batch| B[Bronze: Raw Storage]
    A -->|Stream| C[Kafka]
    C --> B
    B --> D[Data Quality Check]
    D -->|Pass| E[Silver Layer]
    D -->|Fail| F[Quarantine]

    style A fill:#f9f,stroke:#333
    style B fill:#bbf,stroke:#333
    style C fill:#fbb,stroke:#333
    style D fill:#ffb,stroke:#333
    style E fill:#bfb,stroke:#333
    style F fill:#f88,stroke:#333

# src/ingestion/batch_ingest.py
from datetime import datetime
from pathlib import Path
import pandas as pd
from pydantic import BaseModel
from typing import List
import json

class BatchConfig(BaseModel):
    source_uri: str
    target_dir: Path
    file_format: str = "parquet"
    partition_by: List[str] = ["ingest_date"]

def ingest_batch(config: BatchConfig) -> str:
    """Ingest batch data with validation and partitioning."""
    df = pd.read_json(config.source_uri, lines=True)

    # Add metadata
    df["ingest_timestamp"] = datetime.utcnow()
    df["ingest_date"] = df["ingest_timestamp"].dt.date

    # Save with partitioning
    output_path = config.target_dir / f"ingest_{int(datetime.now().timestamp())}"
    df.to_parquet(
        output_path,
        partition_cols=config.partition_by,
        compression="snappy"
    )
    return str(output_path)

# src/ingestion/stream_ingest.py
from kafka import KafkaConsumer
import json
from pathlib import Path
from datetime import datetime
import pandas as pd

class StreamConfig:
    def __init__(self, bootstrap_servers: str, topic: str, batch_size: int = 100):
        self.bootstrap_servers = bootstrap_servers
        self.topic = topic
        self.batch_size = batch_size

def consume_messages(config: StreamConfig, output_dir: Path):
    """Consume messages from Kafka and save in micro-batches."""
    consumer = KafkaConsumer(
        config.topic,
        bootstrap_servers=config.bootstrap_servers,
        value_deserializer=lambda x: json.loads(x.decode('utf-8'))
    )

    batch = []
    for message in consumer:
        batch.append(message.value)

        if len(batch) >= config.batch_size:
            process_batch(batch, output_dir)
            batch = []

def process_batch(messages: list, output_dir: Path):
    """Process and save a micro-batch of messages."""
    df = pd.DataFrame(messages)
    timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
    output_path = output_dir / f"stream_batch_{timestamp}.parquet"
    df.to_parquet(output_path)

# great_expectations/expectations/bronze_suite.json
{
  "data_asset_type": "Dataset",
  "expectation_suite_name": "bronze_validation",
  "expectations": [
    {
      "expectation_type": "expect_column_to_exist",
      "kwargs": {"column": "transcription"}
    },
    {
      "expectation_type": "expect_column_values_to_not_be_null",
      "kwargs": {"column": "medical_specialty"}
    },
    {
      "expectation_type": "expect_column_values_to_match_regex",
      "kwargs": {
        "column": "medical_specialty",
        "regex": "^[A-Za-z\s]+$"
      }
    }
  ]
}

# docker-compose.yml
version: '3.8'

services:
  spark:
    image: bitnami/spark:3.4
    volumes:
      - ./src:/app
    command: >
      spark-submit --master local[*]
      --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.4.0
      /app/stream_ingest.py

  kafka:
    image: bitnami/kafka:3.4
    ports:
      - "9092:9092"
    environment:
      - KAFKA_CFG_NODE_ID=0
      - KAFKA_CFG_PROCESS_ROLES=controller,broker
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092

  minio:
    image: minio/minio
    ports:
      - "9000:9000"
      - "9001:9001"
    environment:
      - MINIO_ROOT_USER=minioadmin
      - MINIO_ROOT_PASSWORD=minioadmin
    command: server /data --console-address ":9001"

from prometheus_client import start_http_server, Counter, Histogram
import time

# Metrics
INGESTED_RECORDS = Counter(
    'ingested_records_total',
    'Total number of ingested records',
    ['source', 'status']
)

PROCESSING_TIME = Histogram(
    'ingestion_processing_seconds',
    'Time spent processing ingestion batches',
    ['source']
)

def process_with_metrics(source: str, batch: list):
    start_time = time.time()
    try:
        # Processing logic here
        INGESTED_RECORDS.labels(source=source, status='success').inc(len(batch))
    except Exception:
        INGESTED_RECORDS.labels(source=source, status='error').inc(len(batch))
        raise
    finally:
        PROCESSING_TIME.labels(source=source).observe(time.time() - start_time)

# Start metrics server
start_http_server(8000)