data-engineering-command-center

# 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 |