Transform raw data into analytical assets using ETL/ELT patterns, SQL (dbt), Python (pandas/polars/PySpark), and orchestration (Airflow). Use when building data pipelines, implementing incremental models, migrating from pandas to polars, or orchestrating multi-step transformations with testing and quality checks.
npxskills add vuralserhat86/antigravity-agentic-skills--skill data_transformLoading…
Transform raw data into analytical assets using modern transformation patterns, frameworks, and orchestration tools.
Select and implement data transformation patterns across the modern data stack. Transform raw data into clean, tested, and documented analytical datasets using SQL (dbt), Python DataFrames (pandas, polars, PySpark), and pipeline orchestration (Airflow, Dagster, Prefect).
Invoke this skill when:
{{
config(
materialized='incremental',
unique_key='order_id'
)
}}
select order_id, customer_id, order_created_at, sum(revenue) as total_revenue
from {{ ref('int_order_items_joined') }}
group by 1, 2, 3
{% if is_incremental() %}
where order_created_at > (select max(order_created_at) from {{ this }})
{% endif %}
import polars as pl
result = (
pl.scan_csv('large_dataset.csv')
.filter(pl.col('year') == 2024)
.with_columns([(pl.col('quantity') * pl.col('price')).alias('revenue')])
.group_by('region')
.agg(pl.col('revenue').sum())
.collect() # Execute lazy query
)
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
with DAG(
dag_id='daily_sales_pipeline',
schedule_interval='0 2 * * *',
default_args={'retries': 2, 'retry_delay': timedelta(minutes=5)},
start_date=datetime(2024, 1, 1),
catchup=False
) as dag:
extract = PythonOperator(task_id='extract', python_callable=extract_data)
transform = PythonOperator(task_id='transform', python_callable=transform_data)
extract >> transform
Use ELT (Extract, Load, Transform) when:
Tools: dbt, Dataform, Snowflake tasks, BigQuery scheduled queries
Use ETL (Extract, Transform, Load) when:
Tools: AWS Glue, Azure Data Factory, custom Python scripts
Use Hybrid when combining sensitive data cleansing (ETL) with analytics transformations (ELT).
Default recommendation: ELT with dbt unless specific compliance or performance constraints require ETL.
For detailed patterns, see references/etl-vs-elt-patterns.md.
Choose pandas when:
Choose polars when:
Choose PySpark when:
Migration path: pandas → polars (easier, similar API) or pandas → PySpark (requires cluster)
For comparisons and migration guides, see references/dataframe-comparison.md.
Choose Airflow when:
Choose Dagster when:
dbt_assets integration)Choose Prefect when:
Safe default: Airflow (battle-tested) unless specific needs for Dagster/Prefect.
For detailed patterns, see references/orchestration-patterns.md.
Staging Layer (models/staging/)
Intermediate Layer (models/intermediate/)
Marts Layer (models/marts/)
View: Query re-run each time model referenced. Use for fast queries, staging layer.
Table: Full refresh on each run. Use for frequently queried models, expensive computations.
Incremental: Only processes new/changed records. Use for large fact tables, event logs.
Ephemeral: CTE only, not persisted. Use for intermediate calculations.
models:
- name: fct_orders
columns:
- name: order_id
tests:
- unique
- not_null
- name: customer_id
tests:
- relationships:
to: ref('dim_customers')
field: customer_id
- name: total_revenue
tests:
- dbt_utils.accepted_range:
min_value: 0
For comprehensive dbt patterns, see:
references/dbt-best-practices.mdreferences/incremental-strategies.mdimport pandas as pd
df = pd.read_csv('sales.csv')
result = (
df
.query('year == 2024')
.assign(revenue=lambda x: x['quantity'] * x['price'])
.groupby('region')
.agg({'revenue': ['sum', 'mean']})
)
import polars as pl
result = (
pl.scan_csv('sales.csv') # Lazy evaluation
.filter(pl.col('year') == 2024)
.with_columns([(pl.col('quantity') * pl.col('price')).alias('revenue')])
.group_by('region')
.agg([
pl.col('revenue').sum().alias('revenue_sum'),
pl.col('revenue').mean().alias('revenue_mean')
])
.collect() # Execute lazy query
)
Key differences:
scan_csv() (lazy) vs pandas read_csv() (eager)with_columns() vs pandas assign()pl.col() expressions vs pandas string referencescollect() to execute lazy queriesfrom pyspark.sql import SparkSession, functions as F
spark = SparkSession.builder.appName("Transform").getOrCreate()
df = spark.read.csv('sales.csv', header=True, inferSchema=True)
result = (
df
.filter(F.col('year') == 2024)
.withColumn('revenue', F.col('quantity') * F.col('price'))
.groupBy('region')
.agg(F.sum('revenue').alias('total_revenue'))
)
For migration guides, see references/dataframe-comparison.md.
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
default_args = {
'owner': 'data-engineering',
'retries': 2,
'retry_delay': timedelta(minutes=5)
}
with DAG(
dag_id='data_pipeline',
default_args=default_args,
schedule_interval='0 2 * * *', # Daily at 2 AM
start_date=datetime(2024, 1, 1),
catchup=False
) as dag:
task1 = PythonOperator(task_id='extract', python_callable=extract_fn)
task2 = PythonOperator(task_id='transform', python_callable=transform_fn)
task1 >> task2 # Define dependency
Linear: A >> B >> C (sequential)
Fan-out: A >> [B, C, D] (parallel after A)
Fan-in: [A, B, C] >> D (D waits for all)
For Airflow, Dagster, and Prefect patterns, see references/orchestration-patterns.md.
Generic tests (reusable): unique, not_null, accepted_values, relationships
Singular tests (custom SQL):
-- tests/assert_positive_revenue.sql
select * from {{ ref('fct_orders') }}
where total_revenue < 0
import great_expectations as gx
context = gx.get_context()
suite = context.add_expectation_suite("orders_suite")
suite.add_expectation(
gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id")
)
suite.add_expectation(
gx.expectations.ExpectColumnValuesToBeBetween(
column="total_revenue", min_value=0
)
)
For comprehensive testing patterns, see references/data-quality-testing.md.
Window functions for analytics:
select
order_date,
daily_revenue,
avg(daily_revenue) over (
partition by region
order by order_date
rows between 6 preceding and current row
) as revenue_7d_ma,
sum(daily_revenue) over (
partition by region
order by order_date
) as cumulative_revenue
from daily_sales
For advanced window functions, see references/window-functions-guide.md.
Ensure transformations produce same result when run multiple times:
merge statements in incremental modelsunique_key in dbt incremental models{% if is_incremental() %}
where created_at > (select max(created_at) from {{ this }})
{% endif %}
try:
result = perform_transformation()
validate_result(result)
except ValidationError as e:
log_error(e)
raise
SQL Transformations: dbt Core (industry standard, multi-warehouse, rich ecosystem)
pip install dbt-core dbt-snowflake
Python DataFrames: polars (10-100x faster than pandas, multi-threaded, lazy evaluation)
pip install polars
Orchestration: Apache Airflow (battle-tested at scale, 5,000+ integrations)
pip install apache-airflow
Working examples in:
examples/python/pandas-basics.py - pandas transformationsexamples/python/polars-migration.py - pandas to polars migrationexamples/python/pyspark-transformations.py - PySpark operationsexamples/python/airflow-data-pipeline.py - Complete Airflow DAGexamples/sql/dbt-staging-model.sql - dbt staging layerexamples/sql/dbt-intermediate-model.sql - dbt intermediate layerexamples/sql/dbt-incremental-model.sql - Incremental patternsexamples/sql/window-functions.sql - Advanced SQLscripts/generate_dbt_models.py - Generate dbt model boilerplatescripts/benchmark_dataframes.py - Compare pandas vs polars performanceFor data ingestion patterns, see ingesting-data.
For data visualization, see visualizing-data.
For database design, see databases-* skills.
For real-time streaming, see streaming-data.
For data platform architecture, see ai-data-engineering.
For monitoring pipelines, see observability.
Design robust ETL/ELT pipelines for data processing.
Use Airflow for orchestration, implement idempotent operations, add error handling, monitor pipeline health.
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
default_args = {
'owner': 'data-team',
'retries': 3,
'retry_delay': timedelta(minutes=5),
'email_on_failure': True,
'email': ['[email protected]']
}
with DAG(
'etl_pipeline',
default_args=default_args,
schedule_interval='0 2 * * *', # Daily at 2 AM
start_date=datetime(2024, 1, 1),
catchup=False
) as dag:
extract = PythonOperator(
task_id='extract_data',
python_callable=extract_from_source
)
transform = PythonOperator(
task_id='transform_data',
python_callable=transform_data
)
load = PythonOperator(
task_id='load_to_warehouse',
python_callable=load_to_warehouse
)
extract >> transform >> load
def extract_incremental(last_run_date):
query = f"""
SELECT * FROM source_table
WHERE updated_at > '{last_run_date}'
"""
return pd.read_sql(query, conn)
def safe_transform(data):
try:
transformed = transform_data(data)
return transformed
except Exception as e:
logger.error(f"Transform failed: {e}")
send_alert(f"Pipeline failed: {e}")
raise
Kaynak: dbt Labs - Best Practices & Polars Performance Guide
dbt-core profilini kur veya Cloud IDE yapılandır.lazy modunu (scan_csv / collect) kullanarak bellek ve hız optimizasyonu yap.dbt tests veya Great Expectations ile validation yaz.| Aşama | Doğrulama |
|---|---|
| 1 | Dönüşüm sonrası veri kaybı yaşandı mı? (Check Sum) |
| 2 | dbt modellerinde ref fonksiyonu dışında hardcoded tablo ismi kullanıldı mı? |
| 3 | Pipeline başarısız olduğunda "Rollback" veya "Reprocessing" stratejisi var mı? |
Data Transformation v2.0 - With Workflow
Create or update AgentSkills. Use when designing, structuring, or packaging skills with scripts, references, and assets.
Create or update AgentSkills. Use when designing, structuring, or packaging skills with scripts, references, and assets.
Set up and use 1Password CLI (op). Use when installing the CLI, enabling desktop app integration, signing in (single or multi-account), or reading/injecting/running secrets via op.
CLI to manage emails via IMAP/SMTP. Use `himalaya` to list, read, write, reply, forward, search, and organize emails from the terminal. Supports multiple accounts and message composition with MML (MIME Meta Language).
Create or update AgentSkills. Use when designing, structuring, or packaging skills with scripts, references, and assets.
Create or update AgentSkills. Use when designing, structuring, or packaging skills with scripts, references, and assets.
Set up and use 1Password CLI (op). Use when installing the CLI, enabling desktop app integration, signing in (single or multi-account), or reading/injecting/running secrets via op.
CLI to manage emails via IMAP/SMTP. Use `himalaya` to list, read, write, reply, forward, search, and organize emails from the terminal. Supports multiple accounts and message composition with MML (MIME Meta Language).