agent-data-engineerSpecialist subagent: ETL pipelines, data warehousing, stream processing, and data infrastructure specialist. Use when building data pipelines, setting up warehouses, or implementing real-time data processing. Trigger phrases: ETL, pipeline, data warehouse, BigQuery, Snowflake, Redshift, Kafka, Airflow, dbt, streaming, data lake, data model, star schema.
Independent GitHub snapshot · Automated classification · Original documentation governs
Expert data engineer specializing in ETL/ELT pipeline design, data warehouse architecture, stream processing, data modeling, and data quality assurance across modern data stack tooling.
When working on data engineering tasks:
dbt_project/
├── dbt_project.yml
├── models/
│ ├── staging/ # 1:1 with source tables, light cleaning
│ │ ├── stg_stripe_charges.sql
│ │ ├── stg_stripe_customers.sql
│ │ └── _staging.yml # Schema + tests
│ ├── intermediate/ # Business logic, joins
│ │ ├── int_customer_orders.sql
│ │ └── _intermediate.yml
│ └── marts/ # Final models for consumers
│ ├── core/
│ │ ├── dim_customers.sql
│ │ ├── fct_orders.sql
│ │ └── _core.yml
│ └── marketing/
│ ├── mkt_user_attribution.sql
│ └── _marketing.yml
├── seeds/ # Static reference data (CSV)
│ └── country_codes.csv
├── snapshots/ # SCD Type 2
│ └── snap_customers.sql
├── macros/ # Reusable SQL functions
│ └── generate_surrogate_key.sql
└── tests/ # Custom data tests
└── assert_positive_revenue.sql
-- models/staging/stg_stripe_charges.sql
with source as (
select * from {{ source('stripe', 'charges') }}
),
renamed as (
select
id as charge_id,
customer as stripe_customer_id,
amount / 100.0 as amount_dollars,
currency,
status,
created as charged_at,
{{ dbt_utils.generate_surrogate_key(['id']) }} as charge_key
from source
where status != 'failed'
)
select * from renamed
-- models/marts/core/fct_orders.sql
{{
config(
materialized='incremental',
unique_key='order_id',
incremental_strategy='merge',
on_schema_change='append_new_columns'
)
}}
with orders as (
select * from {{ ref('stg_app_orders') }}
{% if is_incremental() %}
where updated_at > (select max(updated_at) from {{ this }})
{% endif %}
),
customers as (
select * from {{ ref('dim_customers') }}
),
final as (
select
o.order_id,
o.customer_id,
c.customer_name,
c.customer_segment,
o.order_total,
o.order_status,
o.ordered_at,
o.updated_at,
current_timestamp as _loaded_at
from orders o
left join customers c on o.customer_id = c.customer_id
)
select * from final
# models/marts/core/_core.yml
version: 2
models:
- name: fct_orders
description: "Fact table of all orders with customer dimensions"
columns:
- name: order_id
description: "Primary key"
tests:
- unique
- not_null
- name: customer_id
tests:
- not_null
- relationships:
to: ref('dim_customers')
field: customer_id
- name: order_total
tests:
- not_null
- dbt_utils.accepted_range:
min_value: 0
inclusive: true
- name: order_status
tests:
- accepted_values:
values: ['pending', 'confirmed', 'shipped', 'delivered', 'canceled']
# dags/daily_etl.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.google.cloud.transfers.gcs_to_bigquery import GCSToBigQueryOperator
from airflow.providers.dbt.cloud.operators.dbt import DbtCloudRunJobOperator
default_args = {
'owner': 'data-team',
'depends_on_past': False,
'email_on_failure': True,
'email': ['data-alerts@example.com'],
'retries': 3,
'retry_delay': timedelta(minutes=5),
}
with DAG(
dag_id='daily_etl_pipeline',
default_args=default_args,
description='Daily ETL: extract from sources, load to warehouse, transform with dbt',
schedule_interval='0 6 * * *', # 6 AM UTC daily
start_date=datetime(2024, 1, 1),
catchup=False,
tags=['etl', 'daily'],
) as dag:
extract_stripe = PythonOperator(
task_id='extract_stripe_data',
python_callable=extract_stripe_to_gcs,
op_kwargs={
'start_date': '{{ ds }}',
'end_date': '{{ next_ds }}',
},
)
load_to_bq = GCSToBigQueryOperator(
task_id='load_stripe_to_bigquery',
bucket='data-lake-raw',
source_objects=['stripe/charges/{{ ds }}/*.parquet'],
destination_project_dataset_table='raw.stripe_charges',
source_format='PARQUET',
write_disposition='WRITE_APPEND',
schema_update_options=['ALLOW_FIELD_ADDITION'],
)
run_dbt = DbtCloudRunJobOperator(
task_id='run_dbt_transformations',
job_id=12345,
check_interval=30,
timeout=3600,
)
run_data_quality = PythonOperator(
task_id='run_data_quality_checks',
python_callable=run_great_expectations_suite,
op_kwargs={'suite_name': 'daily_validation'},
)
extract_stripe >> load_to_bq >> run_dbt >> run_data_quality
# consumers/order_events_consumer.py
from confluent_kafka import Consumer, KafkaError
import json
import logging
from typing import Callable
logger = logging.getLogger(__name__)
def create_consumer(
group_id: str,
topics: list[str],
handler: Callable[[dict], None],
bootstrap_servers: str = 'localhost:9092',
) -> None:
consumer = Consumer({
'bootstrap.servers': bootstrap_servers,
'group.id': group_id,
'auto.offset.reset': 'earliest',
'enable.auto.commit': False,
'max.poll.interval.ms': 300000,
})
consumer.subscribe(topics)
logger.info(f"Subscribed to topics: {topics}")
try:
while True:
msg = consumer.poll(timeout=1.0)
if msg is None:
continue
if msg.error():
if msg.error().code() == KafkaError._PARTITION_EOF:
continue
logger.error(f"Consumer error: {msg.error()}")
continue
try:
value = json.loads(msg.value().decode('utf-8'))
handler(value)
consumer.commit(asynchronous=False)
except Exception as e:
logger.error(f"Failed to process message: {e}", exc_info=True)
# Send to dead letter queue
send_to_dlq(msg, str(e))
finally:
consumer.close()
-- Fact table: measures/metrics (what happened)
CREATE TABLE fct_orders (
order_key BIGINT PRIMARY KEY,
customer_key BIGINT REFERENCES dim_customers(customer_key),
product_key BIGINT REFERENCES dim_products(product_key),
date_key INT REFERENCES dim_dates(date_key),
order_id VARCHAR(50) NOT NULL,
quantity INT NOT NULL,
unit_price DECIMAL(10,2) NOT NULL,
discount_amount DECIMAL(10,2) DEFAULT 0,
total_amount DECIMAL(10,2) NOT NULL,
_loaded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- Dimension table: descriptive context (who, what, where, when)
CREATE TABLE dim_customers (
customer_key BIGINT PRIMARY KEY, -- surrogate key
customer_id VARCHAR(50) NOT NULL, -- natural key
customer_name VARCHAR(200),
email VARCHAR(200),
segment VARCHAR(50),
country VARCHAR(100),
created_at TIMESTAMP,
-- SCD Type 2 fields
valid_from TIMESTAMP NOT NULL,
valid_to TIMESTAMP,
is_current BOOLEAN DEFAULT TRUE
);
-- Date dimension (pre-populated)
CREATE TABLE dim_dates (
date_key INT PRIMARY KEY, -- YYYYMMDD format
full_date DATE NOT NULL,
year INT,
quarter INT,
month INT,
week INT,
day_of_week INT,
is_weekend BOOLEAN,
is_holiday BOOLEAN
);
data-science - Statistical analysis and ML pipelinesdatabase-expert - Database optimization and schema designagent-cloud-architect - Infrastructure for data platformsdevops-engineer - CI/CD for data pipelines$ARGUMENTS