ETL Best Practices

1-- Bronze layer (raw data)
2CREATE TABLE bronze.orders AS
3SELECT * FROM source.raw_orders;
4
5-- Silver layer (cleaned data)
6CREATE TABLE silver.orders AS
7SELECT 
8  order_id,
9  customer_id,
10  COALESCE(order_date, CURRENT_DATE) as order_date,
11  ROUND(total_amount, 2) as total_amount
12FROM bronze.orders
13WHERE order_id IS NOT NULL;
14
15-- Gold layer (business aggregates)
16CREATE TABLE gold.daily_sales AS
17SELECT 
18  order_date,
19  COUNT(*) as order_count,
20  SUM(total_amount) as total_revenue
21FROM silver.orders
22GROUP BY order_date;

1-- Uložení watermarku poslední extrakce
2CREATE TABLE etl.watermarks (
3  table_name VARCHAR(255) PRIMARY KEY,
4  last_extracted_at TIMESTAMP
5);
6
7-- Inkrementální extrakce
8SELECT *
9FROM source.orders
10WHERE updated_at > (
11  SELECT last_extracted_at 
12  FROM etl.watermarks 
13  WHERE table_name = 'orders'
14);
15
16-- Aktualizace watermarku po úspěšné extrakci
17UPDATE etl.watermarks 
18SET last_extracted_at = CURRENT_TIMESTAMP
19WHERE table_name = 'orders';

1import requests
2from time import sleep
3from typing import Generator
4
5def extract_from_api(
6    base_url: str,
7    api_key: str,
8    page_size: int = 100
9) -> Generator[dict, None, None]:
10    """
11    Generátor pro paginovanou extrakci z API
12    s automatickým retry a rate limiting.
13    """
14    page = 1
15    retries = 3
16    
17    while True:
18        try:
19            response = requests.get(
20                f"{base_url}?page={page}&limit={page_size}",
21                headers={"Authorization": f"Bearer {api_key}"},
22                timeout=30
23            )
24            response.raise_for_status()
25            
26            data = response.json()
27            if not data.get("items"):
28                break
29                
30            for item in data["items"]:
31                yield item
32            
33            page += 1
34            sleep(0.5)  # Rate limiting
35            
36        except requests.exceptions.RequestException as e:
37            if retries > 0:
38                retries -= 1
39                sleep(2 ** (3 - retries))  # Exponential backoff
40            else:
41                raise

1-- models/marts/customers/customer_metrics.sql
2
3{{ config(
4    materialized='table',
5    schema='marts'
6) }}
7
8WITH orders AS (
9    SELECT * FROM {{ ref('stg_orders') }}
10),
11
12customers AS (
13    SELECT * FROM {{ ref('stg_customers') }}
14),
15
16customer_orders AS (
17    SELECT
18        customer_id,
19        COUNT(*) as total_orders,
20        SUM(order_amount) as lifetime_value,
21        MIN(order_date) as first_order_date,
22        MAX(order_date) as last_order_date
23    FROM orders
24    GROUP BY customer_id
25)
26
27SELECT
28    c.customer_id,
29    c.customer_name,
30    c.email,
31    c.segment,
32    COALESCE(co.total_orders, 0) as total_orders,
33    COALESCE(co.lifetime_value, 0) as lifetime_value,
34    co.first_order_date,
35    co.last_order_date,
36    DATEDIFF('day', co.last_order_date, CURRENT_DATE) as days_since_last_order
37FROM customers c
38LEFT JOIN customer_orders co ON c.customer_id = co.customer_id

1# models/staging/schema.yml
2version: 2
3
4models:
5  - name: stg_orders
6    description: "Staging tabulka objednávek"
7    columns:
8      - name: order_id
9        description: "Primární klíč"
10        tests:
11          - unique
12          - not_null
13      
14      - name: customer_id
15        description: "Cizí klíč na zákazníka"
16        tests:
17          - not_null
18          - relationships:
19              to: ref('stg_customers')
20              field: customer_id
21      
22      - name: order_amount
23        description: "Celková částka objednávky"
24        tests:
25          - not_null
26          - dbt_utils.accepted_range:
27              min_value: 0
28              max_value: 1000000
29      
30      - name: order_status
31        tests:
32          - accepted_values:
33              values: ['pending', 'confirmed', 'shipped', 'delivered', 'cancelled']