安全扫描（ClawHub）

• VirusTotal ：良性
• OpenClaw ：良性

OpenClaw 评估

Instruction-only data-engineering methodology: the skill is a knowledge/template pack and does not request credentials, install code, or perform actions that contradict its stated purpose.

安装（复制给龙虾 AI）

将下方整段复制到龙虾中文库对话中，由龙虾按 SKILL.md 完成安装。

请把本段交给龙虾中文库（龙虾 AI）执行：为本机安装 OpenClaw 技能「Data Engineering」。简介：Design and operate scalable data pipelines and architectures using best-fit pat…。
请 fetch 以下地址读取 SKILL.md 并按文档完成安装：https://raw.githubusercontent.com/openclaw/skills/refs/heads/main/skills/1kalin/afrexai-data-engineering/SKILL.md
（来源：yingzhi8.cn 技能库）

SKILL.md

打开原始 SKILL.md（GitHub raw）

# Data Engineering Command Center

Complete methodology for designing, building, operating, and scaling data pipelines and infrastructure. Zero dependencies — pure agent skill.

---

## Phase 1: Data Architecture Assessment

Before building anything, understand the landscape.

### Architecture Brief

```yaml
project_name: ""
business_context: ""
data_consumers:
  - team: ""
    use_case: ""          # analytics | ML | operational | reporting | reverse-ETL
    latency_requirement: ""  # real-time (<1s) | near-real-time (<5min) | batch (hourly+)
    query_pattern: ""     # ad-hoc | scheduled | API | dashboard

current_state:
  sources: []             # list every system producing data
  storage: []             # where data lives today
  pain_points: []         # what's broken, slow, unreliable
  data_volume:
    current_gb_per_day: 0
    growth_rate_percent: 0
    retention_months: 0

constraints:
  budget_monthly_usd: 0
  team_size: 0
  skill_level: ""         # junior | mid | senior | mixed
  compliance: []          # GDPR, HIPAA, SOX, PCI, none
  cloud_provider: ""      # AWS | GCP | Azure | multi | on-prem
```

### Architecture Pattern Decision Matrix

| Signal | Pattern | When to Use |
|--------|---------|-------------|
| All consumers need data hourly+ | **Batch ETL** | Reporting, warehousing, most analytics |
| Some need <5 min latency | **Micro-batch** | Dashboard freshness, near-real-time analytics |
| Events need <1s processing | **Streaming** | Fraud detection, real-time pricing, alerts |
| Need both batch + streaming | **Lambda** | When batch accuracy + real-time speed both matter |
| Want to simplify Lambda | **Kappa** | When you can reprocess from stream replay |
| Data lake + warehouse combined | **Lakehouse** | When you need both cheap storage + fast SQL |
| Sources change independently | **Data Mesh** | Large orgs, domain-owned data, >5 teams |
| ML is primary consumer | **Feature Store** | ML-heavy orgs with feature reuse needs |

### Technology Selection Guide

#### Orchestration

| Tool | Best For | Avoid When |
|------|----------|------------|
| **Airflow** | Complex DAGs, Python-native teams, mature ecosystem | Simple pipelines (<5 tasks) |
| **Dagster** | Software-defined assets, strong typing, dev experience | Legacy team resistant to new paradigms |
| **Prefect** | Dynamic workflows, cloud-native, Python-first | Need on-prem with no cloud dependency |
| **dbt** | SQL transformations, ELT, analytics engineering | Non-SQL transforms, streaming |
| **Temporal** | Long-running workflows, retry-heavy, microservices | Simple ETL, small teams |
| **Cron + scripts** | <3 pipelines, solo engineer, simple schedules | Anything with dependencies or retries |

#### Processing

| Tool | Best For | Avoid When |
|------|----------|------------|
| **Spark** | >100GB, complex transforms, ML pipelines | <10GB (overkill), real-time streaming |
| **DuckDB** | Local analytics, <100GB, SQL on files | Distributed processing, production streaming |
| **Polars** | Single-node, Rust-speed, <50GB, DataFrames | Distributed, need Spark ecosystem |
| **Pandas** | <1GB, quick analysis, prototyping | Production pipelines, anything >5GB |
| **Flink** | True streaming, event-time processing | Batch-only, small team (steep learning curve) |
| **SQL (warehouse)** | ELT in Snowflake/BigQuery/Redshift | Complex ML transforms, binary data |

#### Storage

| Tool | Best For | Avoid When |
|------|----------|------------|
| **Snowflake** | Analytics, separation of compute/storage, multi-cloud | Tight budget, real-time OLTP |
| **BigQuery** | GCP-native, serverless, large-scale analytics | Multi-cloud, need fine-grained cost control |
| **Redshift** | AWS-native, existing AWS ecosystem | Elastic scaling needs, multi-cloud |
| **Databricks** | ML + analytics unified, Spark-native, lakehouse | Pure SQL analytics, small data |
| **PostgreSQL** | OLTP + light analytics, <500GB, budget-conscious | >1TB analytics, real-time dashboards on large data |
| **S3/GCS/ADLS** | Raw data lake, cheap storage, any format | Direct SQL queries (need compute layer) |
| **Delta Lake/Iceberg** | Table format on data lake, ACID on files | Simple file storage, no lakehouse need |

---

## Phase 2: Data Modeling

### Modeling Methodology Decision

| Approach | Best For | Key Concept |
|----------|----------|-------------|
| **Kimball (Dimensional)** | BI/reporting, star schemas | Facts + Dimensions, business-process-centric |
| **Inmon (3NF)** | Enterprise data warehouse, single source of truth | Normalized, subject-area-centric |
| **Data Vault 2.0** | Agile warehousing, auditability, multiple sources | Hubs + Links + Satellites, insert-only |
| **One Big Table (OBT)** | Simple analytics, few joins, dashboard performance | Pre-joined, denormalized, fast queries |
| **Activity Schema** | Event analytics, product analytics | Entity + Activity + Feature columns |

### Dimensional Model Template

```yaml
fact_table:
  name: "fact_[business_process]"
  grain: ""                    # one row = one [what]?
  grain_statement: "One row per [transaction/event/snapshot] at [time grain]"
  measures:
    - name: ""
      type: ""                 # additive | semi-additive | non-additive
      aggregation: ""          # SUM | AVG | COUNT | MIN | MAX | COUNT DISTINCT
      business_definition: ""
  degenerate_dimensions: []    # IDs stored in fact (order_number, invoice_id)
  foreign_keys: []             # links to dimension tables

dimensions:
  - name: "dim_[entity]"
    type: ""                   # Type 1 (overwrite) | Type 2 (history) | Type 3 (previous value)
    natural_key: ""            # business key from source
    surrogate_key: ""          # warehouse-generated key
    attributes:
      - name: ""
        source: ""
        scd_type: ""           # 1 | 2 | 3
    hierarchy: []              # e.g., [country, region, city, store]
```

### SCD Type Decision Guide

| Scenario | SCD Type | Implementation |
|----------|----------|----------------|
| Don't care about history | **Type 1** | UPDATE in place |
| Need full history | **Type 2** | New row + valid_from/valid_to + is_current flag |
| Only need previous value | **Type 3** | Add previous_[column] |
| Track changes with timestamps | **Type 4** | Mini-dimension (history table) |
| Hybrid: some attrs Type 1, some Type 2 | **Type 6** | Combine 1+2+3 in one table |

**Default recommendation:** Type 2 for anything business-critical (customer status, product price, employee department). Type 1 for everything else.

### Naming Conventions

| Object | Convention | Example |
|--------|-----------|---------|
| Raw/staging tables | `raw_[source]_[table]` | `raw_stripe_payments` |
| Staging models | `stg_[source]__[entity]` | `stg_stripe__payments` |
| Intermediate models | `int_[entity]_[verb]` | `int_orders_pivoted` |
| Mart/fact tables | `fct_[business_process]` | `fct_orders` |
| Dimension tables | `dim_[entity]` | `dim_customers` |
| Metrics/aggregates | `mrt_[domain]_[metric]` | `mrt_sales_daily` |
| Snapshots | `snp_[entity]_[grain]` | `snp_inventory_daily` |
| Columns: boolean | `is_[state]` or `has_[thing]` | `is_active`, `has_subscription` |
| Columns: timestamp | `[event]_at` | `created_at`, `shipped_at` |
| Columns: date | `[event]_date` | `order_date` |
| Columns: ID | `[entity]_id` | `customer_id` |
| Columns: amount | `[thing]_amount` | `order_amount` |
| Columns: count | `[thing]_count` | `line_item_count` |

---

## Phase 3: Pipeline Design Patterns

### Universal Pipeline Template

```yaml
pipeline:
  name: ""
  owner: ""
  schedule: ""               # cron expression
  sla_minutes: 0             # max acceptable runtime
  tier: ""                   # 1 (critical) | 2 (important) | 3 (nice-to-have)

  extract:
    source_system: ""
    connection: ""
    strategy: ""             # full | incremental | CDC | log-based
    incremental_key: ""      # column for incremental (e.g., updated_at)
    watermark_storage: ""    # where to persist last-extracted position

  transform:
    engine: ""               # SQL | Spark | Python | dbt
    stages:
      - name: "clean"
        operations: []       # dedupe, null handling, type casting, trimming
      - name: "conform"
        operations: []       # standardize codes, currencies, timezones
      - name: "enrich"
        operations: []       # lookups, calculations, derived fields
      - name: "aggregate"
        operations: []       # rollups, pivots, window functions

  load:
    target_system: ""
    strategy: ""             # append | upsert | merge | truncate-reload | partition-swap
    merge_keys: []
    partition_key: ""
    clustering_keys: []

  quality_gates:
    pre_load: []             # checks before writing
    post_load: []            # checks after writing

  error_handling:
    strategy: ""             # fail-fast | dead-letter | retry | skip-and-alert
    max_retries: 3
    retry_delay_seconds: 300
    alert_channels: []
```

### Extraction Strategy Decision Tree

```
Is the source database?
├── Yes → Does it support CDC?
│   ├── Yes → Use CDC (Debezium, AWS DMS, Fivetran)
│   │   Best for: high-volume, low-latency, minimal source impact
│   └── No → Does it have a reliable updated_at column?
│       ├── Yes → Incremental extraction on updated_at
│       │   ⚠️ Won't catch hard deletes — add periodic full reconciliation
│       └── No → Full extraction
│           Only viable for small tables (<1M rows)
├── Is it an API?
│   ├── Supports webhooks? → Event-driven ingestion
│   ├── Has cursor/pagination? → Incremental with cursor bookmark
│   └── No pagination? → Full pull with rate-limit handling
├── Is it files (S3, SFTP, email)?
│   └── Event-triggered (S3 notification, file watcher)
│       Validate: schema, completeness, filename pattern
└── Is it streaming (Kafka, Kinesis, Pub/Sub)?
    └── Consumer group with offset management
        Key decisions: at-least-once vs exactly-once, consumer lag alerting
```

### Load Strategy Decision

| Strategy | When | Trade-off |
|----------|------|-----------|
| **Append** | Event/log data, immutable facts | Simple but grows forever — partition + retain |
| **Upsert/Merge** | Dimension updates, SCD Type 1 | Handles updates but slower on large tables |
| **Truncate-Reload** | Small tables (<1M), reference data | Simple but window of missing data |
| **Partition Swap** | Large fact tables, daily loads | Atomic, fast, but needs partition alignment |
| **Soft Delete** | Need audit trail of deletions | Adds complexity to every downstream query |

### Idempotency Rules (NON-NEGOTIABLE)

Every pipeline MUST be re-runnable without side effects:

1. **Use MERGE/UPSERT, never blind INSERT** for mutable data
2. **Partition-swap for immutable data** — drop partition + reload
3. **Store watermarks externally** — not in the pipeline code
4. **Dedup at ingestion** — use source natural keys
5. **Test by running twice** — output must be identical both times

---

## Phase 4: Data Quality Framework

### Quality Dimensions

| Dimension | Definition | Example Check |
|-----------|-----------|---------------|
| **Completeness** | No missing values where required | `NOT NULL` on required fields, row count within range |
| **Uniqueness** | No unexpected duplicates | Primary key uniqueness, natural key uniqueness |
| **Validity** | Values within expected domain | Enum checks, range checks, regex patterns |
| **Accuracy** | Data matches real-world truth | Cross-system reconciliation, manual spot checks |
| **Freshness** | Data arrives on time | `MAX(loaded_at) > NOW() - INTERVAL '2 hours'` |
| **Consistency** | Same data agrees across systems | Sum reconciliation between source and target |

### Quality Check Templates

```sql
-- Completeness: Required fields not null
SELECT COUNT(*) AS null_violations
FROM {table}
WHERE {required_column} IS NULL;
-- Threshold: 0

-- Uniqueness: No duplicate primary keys
SELECT {pk_column}, COUNT(*) AS dupe_count
FROM {table}
GROUP BY {pk_column}
HAVING COUNT(*) > 1;
-- Threshold: 0

-- Freshness: Data arrived within SLA
SELECT CASE
  WHEN MAX({timestamp_col}) > CURRENT_TIMESTAMP - INTERVAL '{sla_hours} hours'
  THEN 'PASS' ELSE 'FAIL'
END AS freshness_check
FROM {table};

-- Volume: Row count within expected range
SELECT CASE
  WHEN COUNT(*) BETWEEN {min_expected} AND {max_expected}
  THEN 'PASS' ELSE 'FAIL'
END AS volume_check
FROM {table}
WHERE {partition_col} = '{run_date}';

-- Referential: FK integrity
SELECT COUNT(*) AS orphan_count
FROM {fact_table} f
LEFT JOIN {dim_table} d ON f.{fk} = d.{pk}
WHERE d.{pk} IS NULL;
-- Threshold: 0

-- Distribution: No unexpected skew
SELECT {column}, COUNT(*) AS cnt,
  ROUND(100.0 * COUNT(*) / SUM(COUNT(*)) OVER (), 2) AS pct
FROM {table}
GROUP BY {column}
ORDER BY cnt DESC;
-- Alert if any single value > {max_pct}%

-- Cross-system reconciliation
SELECT
  (SELECT SUM(amount) FROM source_system.orders WHERE date = '{date}') AS source_total,
  (SELECT SUM(amount) FROM warehouse.fct_orders WHERE order_date = '{date}') AS target_total,
  ABS(source_total - target_total) AS variance;
-- Threshold: variance < 0.01 * source_total (1%)
```

### Data Contract Template

```yaml
contract:
  name: ""
  version: ""
  owner: ""                    # team responsible for producing this data
  consumers: []                # teams consuming this data
  sla:
    freshness_hours: 0
    availability_percent: 99.9
    support_hours: ""          # business-hours | 24x7

  schema:
    - column: ""
      type: ""
      nullable: false
      description: ""
      business_definition: ""
      pii: false
      checks:
        - type: ""             # not_null | unique | range | enum | regex | custom
          params: {}

  breaking_change_policy: ""   # notify-30-days | version-bump | never-break
  notification_channel: ""
```

### Quality Severity Levels

| Level | Definition | Response |
|-------|-----------|----------|
| **P0 — Critical** | Data corruption, wrong numbers in production dashboards, compliance data wrong | Stop pipeline, alert immediately, rollback if possible |
| **P1 — High** | Missing data for key reports, SLA breach, >5% of records affected | Alert team, fix within 4 hours, post-mortem required |
| **P2 — Medium** | Non-critical field quality, <1% records affected, no downstream impact | Fix in next sprint, add monitoring to prevent recurrence |
| **P3 — Low** | Cosmetic issues, edge cases, non-critical data | Backlog, fix when convenient |

---

## Phase 5: Performance Optimization

### SQL Optimization Checklist

| Problem | Fix | Impact |
|---------|-----|--------|
| Full table scan | Add/use partition pruning | 10-100x faster |
| Large joins | Pre-aggregate before joining | 5-50x faster |
| SELECT * | Select only needed columns | 2-10x faster (columnar stores) |
| Correlated subquery | Rewrite as JOIN or window function | 10-100x faster |
| DISTINCT on large result | Fix upstream duplication instead | 2-5x faster |
| ORDER BY without LIMIT | Add LIMIT or remove if not needed | Prevents memory spills |
| String operations in WHERE | Pre-compute, use lookup table | Enables index usage |
| Multiple passes over same data | Combine with CASE WHEN + GROUP BY | 2-5x faster |
| NOT IN with NULLs | Use NOT EXISTS or LEFT JOIN IS NULL | Correctness + performance |

### Spark Optimization Guide

| Problem | Solution |
|---------|----------|
| Shuffle-heavy joins | Broadcast small table (`broadcast(df)`) if <100MB |
| Data skew | Salt the skewed key: add random prefix, join on salted key, aggregate |
| Small files | Coalesce output: `.coalesce(target_files)` or use adaptive query execution |
| Too many partitions | `spark.sql.shuffle.partitions` = 2-3x cluster cores |
| OOM errors | Increase `spark.executor.memory`, reduce partition size, spill to disk |
| Slow writes | Use Parquet with snappy, partition by date, avoid small writes |
| Repeated computation | `.cache()` or `.persist()` DataFrames used >1 time |
| Complex transformations | Push down predicates, filter early, select early |

### Partitioning Strategy

| Data Type | Partition Key | Why |
|-----------|--------------|-----|
| Transactional/event | Date (daily or monthly) | Most queries filter by time range |
| Multi-tenant | Tenant ID + date | Isolate tenant queries, time-range pruning |
| Geospatial | Region + date | Regional queries are common |
| Log data | Date + hour | High volume needs finer partitions |
| Reference/dimension | Don't partition | Too small, full scan is fine |

**Rules:**
- Target 100MB-1GB per partition (compressed)
- <10,000 total partitions per table
- Never partition on high-cardinality columns (user_id)
- Always include partition key in WHERE clauses

---

## Phase 6: Data Governance & Cataloging

### Data Classification

| Level | Examples | Controls |
|-------|---------|----------|
| **Public** | Product catalog, published stats | No restrictions |
| **Internal** | Aggregated metrics, non-PII analytics | Auth required, audit logging |
| **Confidential** | Customer PII, financial records, HR data | Encryption, column-level access, masking |
| **Restricted** | SSN, payment cards, health records, passwords | Encryption at rest + transit, tokenization, audit every access, retention limits |

### PII Handling Rules

1. **Identify:** Scan all sources for PII columns (name, email, phone, SSN, DOB, address, IP)
2. **Classify:** Tag each with sensitivity level
3. **Minimize:** Only ingest PII you actually need
4. **Protect:** 
   - Hash or tokenize in staging (SHA-256 with salt for pseudonymization)
   - Dynamic masking for non-privileged users
   - Column-level encryption for restricted data
5. **Retain:** Auto-delete after retention period
6. **Audit:** Log every query touching PII columns
7. **Right to delete:** Build a deletion pipeline that propagates across all derived tables

### Data Catalog Entry Template

```yaml
dataset:
  name: ""
  description: ""
  owner_team: ""
  steward: ""                  # person responsible for quality
  domain: ""                   # sales | marketing | finance | product | engineering
  tier: ""                     # gold (trusted) | silver (cleaned) | bronze (raw)
  
  lineage:
    sources: []                # upstream datasets/systems
    transformations: ""        # brief description of key transforms
    downstream: []             # who consumes this
  
  refresh:
    schedule: ""
    sla_hours: 0
    last_successful_run: ""
  
  quality:
    tests: []                  # list of quality checks
    last_score: 0              # 0-100
    known_issues: []
  
  access:
    classification: ""         # public | internal | confidential | restricted
    pii_columns: []
    access_request_process: "" # how to get access
  
  usage:
    avg_daily_queries: 0
    top_consumers: []
    cost_monthly_usd: 0
```

---

## Phase 7: Pipeline Monitoring & Alerting

### Pipeline Health Dashboard

```yaml
dashboard:
  pipeline_metrics:
    - metric: "Pipeline Success Rate"
      formula: "successful_runs / total_runs * 100"
      target: ">99%"
      alert_threshold: "<95%"

    - metric: "Average Runtime"
      formula: "avg(end_time - start_time) over 7 days"
      target: "<SLA"
      alert_threshold: ">80% of SLA"

    - metric: "Data Freshness"
      formula: "NOW() - MAX(loaded_at)"
      target: "<SLA hours"
      alert_threshold: ">SLA"

    - metric: "Data Volume Variance"
      formula: "abs(today_rows - avg_7d_rows) / avg_7d_rows * 100"
      target: "<20%"
      alert_threshold: ">50%"

    - metric: "Quality Check Pass Rate"
      formula: "passed_checks / total_checks * 100"
      target: "100%"
      alert_threshold: "<95%"

    - metric: "Failed Pipeline Count"
      formula: "count where status = failed in last 24h"
      target: "0"
      alert_threshold: ">0"

    - metric: "Backfill Queue"
      formula: "count of pending backfill requests"
      target: "0"
      alert_threshold: ">5"

    - metric: "Infrastructure Cost"
      formula: "compute + storage + egress"
      target: "<budget"
      alert_threshold: ">110% budget"
```

### Alert Severity

| Severity | Condition | Response Time | Example |
|----------|-----------|---------------|---------|
| **P0** | Revenue/compliance impacting | 15 min | Payment pipeline down, regulatory report delayed |
| **P1** | Business-critical dashboard stale | 1 hour | Executive dashboard >4h stale |
| **P2** | Non-critical pipeline failed | 4 hours | Marketing attribution delayed |
| **P3** | Warning/degradation | Next business day | Pipeline 80% of SLA, minor quality drift |

### Structured Logging Standard

Every pipeline run MUST log:

```json
{
  "pipeline_name": "",
  "run_id": "",
  "started_at": "",
  "completed_at": "",
  "status": "success|failed|partial",
  "stage": "",
  "rows_extracted": 0,
  "rows_transformed": 0,
  "rows_loaded": 0,
  "rows_rejected": 0,
  "quality_checks_passed": 0,
  "quality_checks_failed": 0,
  "duration_seconds": 0,
  "error_message": "",
  "watermark_before": "",
  "watermark_after": ""
}
```

---

## Phase 8: Testing Strategy

### Pipeline Test Pyramid

| Layer | What to Test | How | When |
|-------|-------------|-----|------|
| **Unit** | Individual transforms, business logic | pytest with fixtures, dbt unit tests | Every PR |
| **Integration** | Source connectivity, schema compatibility | Test against staging/dev environment | Daily + PR |
| **Contract** | Schema hasn't changed, data types stable | Schema registry, contract tests | Every pipeline run |
| **Data Quality** | Completeness, uniqueness, freshness, validity | Quality framework checks | Every pipeline run |
| **E2E** | Full pipeline produces correct output | Golden dataset comparison | Weekly + release |
| **Performance** | Runtime within SLA, no regression | Benchmark against historical runs | Weekly |

### dbt Testing Checklist

```yaml
# For every model, define at minimum:
models:
  - name: fct_orders
    columns:
      - name: order_id
        tests:
          - unique
          - not_null
      - name: customer_id
        tests:
          - not_null
          - relationships:
              to: ref('dim_customers')
              field: customer_id
      - name: order_amount
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0
              max_value: 1000000
      - name: order_status
        tests:
          - accepted_values:
              values: ['pending', 'confirmed', 'shipped', 'delivered', 'cancelled']
      - name: ordered_at
        tests:
          - not_null
          - dbt_utils.recency:
              datepart: day
              field: ordered_at
              interval: 2
```

### Backfill Protocol

When you need to reprocess historical data:

1. **Scope:** Define exact date range and affected tables
2. **Impact assessment:** What downstream models/dashboards will be affected?
3. **Communication:** Notify consumers of temporary data inconsistency
4. **Isolation:** Run backfill in separate compute to avoid impacting current pipelines
5. **Validation:** Compare row counts and key metrics pre/post backfill
6. **Execution:** Process in reverse-chronological order (most recent first)
7. **Monitoring:** Watch for resource spikes, duplicate creation
8. **Verification:** Reconcile against source after completion
9. **Documentation:** Log what was backfilled, why, and any anomalies found

---

## Phase 9: Cost Optimization

### Cloud Cost Reduction Strategies

| Strategy | Savings | Effort |
|----------|---------|--------|
| Right-size compute (auto-scaling) | 20-40% | Low |
| Use spot/preemptible instances for batch | 60-80% | Medium |
| Compress data (Parquet + Snappy/Zstd) | 50-80% storage | Low |
| Lifecycle policies (hot → warm → cold → archive) | 40-70% storage | Low |
| Eliminate unused tables/pipelines | 10-30% | Low |
| Optimize query patterns (partition pruning) | 30-60% compute | Medium |
| Reserved capacity for steady-state | 30-50% | Medium |
| Cache expensive queries | 20-50% compute | Medium |

### Cost Allocation Template

```yaml
cost_tracking:
  by_pipeline:
    - pipeline: ""
      compute_monthly_usd: 0
      storage_monthly_usd: 0
      egress_monthly_usd: 0
      total: 0
      cost_per_row: 0        # total / rows_processed
      business_value: ""     # what revenue/decision does this enable?
      roi_justified: true    # is the cost worth it?

  optimization_opportunities:
    - description: ""
      estimated_savings_usd: 0
      effort: ""             # low | medium | high
      priority: 0            # 1 = do now
```

### Cost Red Flags

- Single pipeline >30% of total spend
- Cost per row increasing month-over-month
- Tables with 0 queries in 30 days
- Dev/staging environments running 24/7
- Full table scans on >1TB tables
- Uncompressed data in cloud storage
- Cross-region data transfer

---

## Phase 10: Operational Runbooks

### Pipeline Failure Triage

```
Pipeline failed →
1. Check error message in logs
   ├── Connection timeout → Check source availability, network, credentials
   ├── Schema mismatch → Source schema changed → update extract + notify
   ├── Data quality check failed → Investigate source data, check for anomalies
   ├── Out of memory → Increase resources or optimize query
   ├── Permission denied → Check IAM roles, token expiry
   ├── Duplicate key violation → Check idempotency, investigate source dupes
   └── Timeout (SLA breach) → Check data volume spike, query plan, cluster health

2. Determine impact
   ├── What dashboards/reports are affected?
   ├── What's the data freshness SLA?
   └── Who needs to be notified?

3. Fix
   ├── Transient (network, timeout) → Retry
   ├── Data issue → Fix source data, re-run with quality gate override if safe
   ├── Schema change → Update pipeline, backfill if needed
   └── Infrastructure → Scale up, file ticket with cloud provider

4. Post-fix
   ├── Verify data correctness
   ├── Update runbook with new failure mode
   └── Add monitoring/alerting to catch earlier next time
```

### Schema Change Management

When a source system changes schema:

1. **Detect:** Schema comparison check in extraction pipeline (hash schema, compare to registered)
2. **Classify:**
   - **Additive** (new column): Usually safe — add to pipeline, backfill if needed
   - **Rename**: Map old → new in transform, update downstream
   - **Type change**: Assess compatibility, may need cast or historical rebuild
   - **Column removed**: Critical — breaks queries, need immediate attention
3. **Test:** Run pipeline in dry-run mode with new schema
4. **Deploy:** Update transforms, quality checks, documentation
5. **Communicate:** Notify downstream consumers via data contract channel

### Disaster Recovery

| Scenario | RPO | RTO | Recovery Steps |
|----------|-----|-----|----------------|
| Pipeline code lost | 0 (git) | 1h | Redeploy from git, restore orchestrator state |
| Warehouse data corrupted | Varies | 4h | Restore from Time Travel/snapshot, re-run affected pipelines |
| Source system down | N/A | Wait | Queue extractions, catch up with incremental once restored |
| Cloud region outage | 24h | 8h | Failover to DR region if configured, else wait |
| Credential compromise | 0 | 2h | Rotate all credentials, audit access logs, re-run affected pipelines |

---

## Phase 11: Advanced Patterns

### Slowly Changing Dimension Type 2 (SQL Template)

```sql
-- Merge pattern for SCD Type 2
MERGE INTO dim_customer AS target
USING (
  SELECT * FROM stg_customers
  WHERE updated_at > (SELECT MAX(valid_from) FROM dim_customer)
) AS source
ON target.customer_natural_key = source.customer_id
   AND target.is_current = TRUE

-- Update: close old record
WHEN MATCHED AND (
  target.customer_name != source.name OR
  target.customer_status != source.status
  -- list all Type 2 tracked columns
) THEN UPDATE SET
  is_current = FALSE,
  valid_to = CURRENT_TIMESTAMP

-- Insert: new record (both new customers and changed ones)
WHEN NOT MATCHED THEN INSERT (
  customer_natural_key, customer_name, customer_status,
  valid_from, valid_to, is_current
) VALUES (
  source.customer_id, source.name, source.status,
  CURRENT_TIMESTAMP, '9999-12-31', TRUE
);

-- Then insert new versions of changed records
INSERT INTO dim_customer (
  customer_natural_key, customer_name, customer_status,
  valid_from, valid_to, is_current
)
SELECT customer_id, name, status,
  CURRENT_TIMESTAMP, '9999-12-31', TRUE
FROM stg_customers s
WHERE EXISTS (
  SELECT 1 FROM dim_customer d
  WHERE d.customer_natural_key = s.customer_id
    AND d.is_current = FALSE
    AND d.valid_to = CURRENT_TIMESTAMP
);
```

### CDC with Debezium (Architecture Pattern)

```
Source DB → Debezium Connector → Kafka Topic → 
  ├── Stream processor (Flink/Spark Streaming) → Target DB
  ├── S3 sink connector → Data Lake (raw)
  └── Elasticsearch sink → Search index
```

Key decisions:
- **Topic per table** or **single topic**: Per table (easier routing, independent scaling)
- **Schema registry**: Always use (Confluent Schema Registry or AWS Glue)
- **Serialization**: Avro (compact + schema evolution) or Protobuf (strict + fast)
- **Offset management**: Connector manages; monitor consumer lag

### Feature Store Pattern

```yaml
feature_store:
  entity: "customer"
  entity_key: "customer_id"
  
  features:
    - name: "total_orders_30d"
      description: "Total orders in last 30 days"
      type: "INT"
      source: "fct_orders"
      computation: "batch"      # batch | streaming | on-demand
      freshness: "daily"
      ttl_hours: 48
    
    - name: "avg_order_value_90d"
      description: "Average order value last 90 days"
      type: "FLOAT"
      source: "fct_orders"
      computation: "batch"
      freshness: "daily"
      ttl_hours: 48
    
    - name: "last_login_minutes_ago"
      description: "Minutes since last login event"
      type: "INT"
      source: "events_stream"
      computation: "streaming"
      freshness: "real-time"
      ttl_hours: 1
  
  serving:
    online: true               # low-latency feature serving (Redis/DynamoDB)
    offline: true              # batch feature retrieval for training
    point_in_time_correct: true  # prevent feature leakage in ML training
```

### Data Mesh Principles

If operating at scale (>5 data teams):

1. **Domain ownership**: Each business domain owns its data products (not central data team)
2. **Data as a product**: Treat datasets like products — SLAs, documentation, discoverability
3. **Self-serve platform**: Central team builds the platform, domains build on top
4. **Federated governance**: Standards and interoperability maintained centrally, implementation decentralized

**When NOT to use Data Mesh:**
- <5 data producers/consumers
- Small team (<20 engineers total)
- Single business domain
- Early-stage company (over-engineering)

---

## Quality Scoring Rubric (0-100)

| Dimension | Weight | Scoring |
|-----------|--------|---------|
| **Pipeline Reliability** | 20 | 0=frequent failures, 10=some failures with manual recovery, 20=99.5%+ success rate with auto-retry |
| **Data Quality** | 20 | 0=no checks, 10=basic null/unique checks, 20=comprehensive quality framework with contracts |
| **Performance** | 15 | 0=regularly breaches SLA, 8=meets SLA, 15=well under SLA with optimization |
| **Documentation** | 10 | 0=none, 5=basic README, 10=full catalog entries with lineage and business definitions |
| **Monitoring** | 15 | 0=no alerts, 8=failure alerts only, 15=proactive monitoring with dashboards and anomaly detection |
| **Testing** | 10 | 0=no tests, 5=basic smoke tests, 10=full test pyramid (unit+integration+contract+E2E) |
| **Cost Efficiency** | 10 | 0=no cost tracking, 5=tracked, 10=optimized with ROI justification per pipeline |

**Scoring guide:**
- 0-40: Critical gaps — prioritize pipeline reliability and data quality
- 41-60: Functional but fragile — add monitoring, testing, documentation
- 61-80: Solid — optimize performance, cost, governance
- 81-100: Excellent — maintain, innovate, mentor

---

## Edge Cases & Gotchas

### Timezone Traps
- Store everything in UTC. Convert only at presentation layer
- Event timestamps: use event time, not processing time
- Daylight saving: `TIMESTAMP WITH TIME ZONE`, never `WITHOUT`
- Late-arriving data: watermark strategy + allowed lateness window

### Late-Arriving Data
- Define maximum acceptable lateness per source
- Reprocess affected partitions when late data arrives
- Track late arrival rate as a quality metric
- Consider separate "late data" pipeline that patches in

### Exactly-Once Processing
- True exactly-once is expensive. Most systems need at-least-once + idempotent writes
- Use transaction IDs or natural keys for deduplication
- Kafka: use idempotent producer + transactional consumer
- Database: MERGE/UPSERT on natural key

### Schema Evolution
- **Forward compatible**: New code reads old data (safe to deploy new readers first)
- **Backward compatible**: Old code reads new data (safe to deploy new writers first)
- **Full compatible**: Both directions (safest, most restrictive)
- Use Avro or Protobuf with schema registry for streaming data

### Multi-Tenant Data
- Tenant ID in every table, every query, every log
- Row-level security in warehouse
- Separate compute per tenant (or at least isolation)
- Never join across tenants without explicit business reason
- Tenant-aware backfill (don't rebuild all tenants for one tenant's issue)

### Data Lake Anti-Patterns
- "Data Swamp": ingesting everything with no organization or catalog → only ingest what has a known consumer
- Small files: thousands of <1MB files → compact regularly (target 100MB-1GB)
- No table format: raw Parquet/CSV without Delta/Iceberg → loses ACID, schema evolution, time travel
- No access controls: single bucket, everyone admin → implement IAM per domain/team

---

## Natural Language Commands

Say any of these to activate specific workflows:

1. **"Design a data pipeline for [source] to [target]"** → Full pipeline template with extraction strategy, transforms, load pattern, quality checks
2. **"Model [entity/domain] for analytics"** → Dimensional model with fact/dimension tables, grain, measures, SCD types
3. **"Optimize this query/pipeline"** → Performance analysis with specific recommendations
4. **"Set up data quality for [table/pipeline]"** → Quality framework with checks, contracts, monitoring
5. **"Audit our data infrastructure"** → Full assessment using scoring rubric
6. **"Help with [Spark/Airflow/dbt/Kafka] issue"** → Troubleshooting with technology-specific guidance
7. **"Design a data catalog for our org"** → Catalog template with governance, classification, lineage
8. **"Plan a data migration from [old] to [new]"** → Migration plan with validation, rollback, parallel-run
9. **"Set up monitoring for our pipelines"** → Dashboard template with alerts, logging standards, runbooks
10. **"Review our data costs"** → Cost analysis with optimization strategies and ROI framework
11. **"Handle schema change in [source]"** → Change management protocol with impact assessment
12. **"Backfill [table] for [date range]"** → Backfill protocol with validation and communication plan