Search the docs, videos, blog, apps, teams, skills and release notes.
At a glance
What it is
Expert ETL pipeline design covering extract patterns (CDC, full load, incremental), transform strategies (SCD types, data cleansing), load patterns (upsert, merge), error handling, data quality checks, scheduling, and orchestration tools for building reliable data integration systems.
A skill is a written procedure a bot loads when a job calls for it. This one is a single file, SKILL.md, and the whole file is on this page.
The whole file, exactly as it downloads. Read it before you rely on it.
SKILL.md
430 lines · 1,676 words
ETL (Extract, Transform, Load) is the backbone of data integration. This skill covers the design and implementation of robust, scalable, and maintainable data pipelines that move data reliably between systems. Modern ETL also encompasses ELT (Extract, Load, Transform) where transformations happen inside the target data warehouse.
Full load extracts the entire dataset each time. Simple but expensive.
Expert ETL pipeline design covering extract patterns (CDC, full load, incremental), transform strategies (SCD types, data cleansing), load patterns (upsert, merge), error handling, data quality checks, scheduling, and orchestration tools for building reliable data integration systems.
Nothing. It ships with Murage, which is open source. Your bots run on the AI plan you already have.
Yes. Open it in Settings → Skills to read, edit or duplicate it. Skill Guard scans it again after every edit.
Download the app, connect the AI you already pay for, and tell your Chief of Staff what needs doing. Plan on about ten minutes from install to a working team.
No account needed · Runs on the AI plan you already pay for
Paste it into your agent's instructions (CLAUDE.md, AGENTS.md or a custom GPT), or add it to a bot in Murage.
Ask Ember for the bot, then switch the skill on in its settings.
Step 1
Send your Chief of Staff this message. It proposes the bot and its role first, and creates it when you say go.
Send this to Ember
“Create a bot that uses the ETL Architect skill, and show me the plan first.”
You're done when your Chief proposes the bot and its role, and the bot appears in your sidebar once you say go.
Step 2
You're done when you see the bot’s skills and the library.
Step 3
You're done when the skill shows as on for that bot.
Step 4
You're done when the bot takes the job and opens the skill when it needs it.
Prefer to do it yourself? Choose + in the sidebar, then New Bot.Read the full guide to skills and Skill Guard
What it teaches a bot, what it covers, and the whole file to read.
What it covers
Overview
Extract Patterns
Transform Strategies
Load Patterns
Error Handling
Data Quality Checks
Scheduling and Orchestration
Monitoring and Observability
ETL vs ELT Decision Framework
When to Use
Output Format
Example
Edge Cases
# Full load pattern with checksum for change detection
import hashlib
import pandas as pd
def full_load_with_checksum(source_query, target_table, engine):
"""Full load with row-level checksum to detect actual changes."""
df = pd.read_sql(source_query, engine)
# Add row checksum for downstream deduplication
checksum_cols = df.columns.tolist()
# MD5 used for non-cryptographic row checksum only. Do NOT use MD5 for passwords or security.
df['_row_checksum'] = df[checksum_cols].apply(
lambda row: hashlib.md5(
'|'.join(str(v) for v in row).encode()
# ... (condensed) ...
df.to_sql(f'{target_table}_staging', engine, if_exists='replace', index=False)
with engine.begin() as conn:
conn.execute(f'ALTER TABLE {target_table} RENAME TO {target_table}_old')
conn.execute(f'ALTER TABLE {target_table}_staging RENAME TO {target_table}')
conn.execute(f'DROP TABLE IF EXISTS {target_table}_old')When to use: Reference/dimension tables under 1M rows, initial loads, data without reliable timestamps.
Extract only new or changed records since the last successful run.
# High-water mark incremental extraction
class IncrementalExtractor:
def __init__(self, source_engine, watermark_store):
self.source = source_engine
self.watermarks = watermark_store
def extract(self, table_name, timestamp_col, batch_size=10000):
last_watermark = self.watermarks.get(table_name, '1970-01-01')
query = f"""
SELECT *
FROM {table_name}
WHERE {timestamp_col} > %(watermark)s
ORDER BY {timestamp_col}
# ... (condensed) ...
result = pd.concat(chunks, ignore_index=True)
self.watermarks.set(table_name, current_watermark)
return result
return pd.DataFrame()Pitfalls: Late-arriving data, clock skew between systems, records updated without changing the timestamp.
CDC captures row-level changes (INSERT, UPDATE, DELETE) from the database transaction log.
# Debezium CDC event processing
import json
def process_cdc_event(event):
"""Process a Debezium CDC event from Kafka."""
payload = event['payload']
operation = payload['op'] # c=create, u=update, d=delete, r=read(snapshot)
result = {
'operation': operation,
'timestamp': payload['ts_ms'],
'source_table': payload['source']['table'],
}
# ... (condensed) ...
"transforms": "unwrap",
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"transforms.unwrap.drop.tombstones": "false",
"transforms.unwrap.delete.handling.mode": "rewrite"
}CDC Decision Matrix:
| Factor | Log-based CDC | Trigger-based CDC | Timestamp-based |
|---|---|---|---|
| Performance impact on source | Minimal | Moderate | Minimal |
| Captures deletes | Yes | Yes | No |
| Captures all changes | Yes | Yes | Only latest state |
| Setup complexity | High | Medium | Low |
| Schema change handling | Requires care | Automatic | Automatic |
-- SCD Type 1: Overwrite (no history)
MERGE INTO dim_customer AS target
USING staging_customer AS source
ON target.customer_id = source.customer_id
WHEN MATCHED THEN
UPDATE SET
name = source.name,
email = source.email,
updated_at = CURRENT_TIMESTAMP
WHEN NOT MATCHED THEN
INSERT (customer_id, name, email, created_at, updated_at)
VALUES (source.customer_id, source.name, source.email,
CURRENT_TIMESTAMP, CURRENT_TIMESTAMP);
# ... (condensed) ...
);
-- SCD Type 3: Limited history (previous + current columns)
-- SCD Type 4: Mini-dimension for rapidly changing attributes
-- SCD Type 6: Hybrid (1 + 2 + 3 combined)import re
from typing import Optional
class DataCleanser:
"""Standard data cleansing operations for ETL pipelines."""
@staticmethod
def standardize_phone(phone: Optional[str], country='US') -> Optional[str]:
if not phone:
return None
digits = re.sub(r'[^\d]', '', phone)
if country == 'US':
if len(digits) == 10:
return f'+1{digits}'
# ... (condensed) ...
continue
if jaro_winkler_similarity(keys[i], keys[j]) >= threshold:
duplicates.add(j)
return [r for idx, r in enumerate(records) if idx not in duplicates]-- PostgreSQL upsert with conflict handling
INSERT INTO target_table (id, name, value, updated_at)
SELECT id, name, value, NOW()
FROM staging_table
ON CONFLICT (id) DO UPDATE SET
name = EXCLUDED.name,
value = EXCLUDED.value,
updated_at = EXCLUDED.updated_at
WHERE target_table.name != EXCLUDED.name
OR target_table.value != EXCLUDED.value;
-- The WHERE clause prevents unnecessary row versioning# PostgreSQL COPY for maximum throughput
import io
import csv
def bulk_load_postgres(df, table_name, engine):
"""Load DataFrame using PostgreSQL COPY protocol (~10x faster than INSERT)."""
buffer = io.StringIO()
df.to_csv(buffer, index=False, header=False, sep='\t', na_rep='\\N',
quoting=csv.QUOTE_NONE, escapechar='\\')
buffer.seek(0)
raw_conn = engine.raw_connection()
try:
cursor = raw_conn.cursor()
# ... (condensed) ...
# Batch size recommendations:
# PostgreSQL COPY: unlimited (streaming), best performance
# INSERT ... VALUES: 1000-5000 rows per statement
# Snowflake COPY INTO: stage files 100-250 MB each
# BigQuery: load jobs, max 15 TB per job, prefer Avro/Parquetclass ETLPipeline:
def __init__(self):
self.dead_letter_queue = []
self.processed_count = 0
self.error_count = 0
def process_batch(self, records, transform_fn, max_error_pct=5.0):
"""Process records with dead letter queue for failures."""
results = []
for record in records:
try:
transformed = transform_fn(record)
results.append(transformed)
# ... (condensed) ...
target.write_dead_letters(self.dead_letter_queue)
count = len(self.dead_letter_queue)
self.dead_letter_queue.clear()
return count
return 0# Idempotent load using partition swap
def idempotent_partition_load(df, table, partition_key, partition_value, engine):
"""
Idempotent load: re-running with same inputs produces same result.
Uses DELETE + INSERT within a transaction.
"""
with engine.begin() as conn:
# Delete existing data for this partition
conn.execute(
f"DELETE FROM {table} WHERE {partition_key} = %s",
(partition_value,)
)
# Insert new data
df.to_sql(table, conn, if_exists='append', index=False)
# If this runs twice with same data, result is identicalclass DataQualityChecker:
"""Run data quality checks between pipeline stages."""
def __init__(self):
self.checks = []
self.results = []
def add_check(self, name, check_fn, severity='error'):
self.checks.append({
'name': name,
'fn': check_fn,
'severity': severity # error, warning, info
})
# ... (condensed) ...
max_ts = df[timestamp_col].max()
age = (pd.Timestamp.utcnow() - max_ts).total_seconds() / 3600
passed = age <= max_age_hours
return passed, f'Data age: {age:.1f}h (max: {max_age_hours}h)'
return _checkHow often does source data change?
Real-time -> Streaming (Kafka + Flink/Spark Streaming)
Minutes -> Micro-batch (5-15 min intervals)
Hourly/Daily -> Batch scheduling
What is the SLA for data freshness?
< 1 minute -> CDC + streaming
< 1 hour -> Micro-batch or frequent batch
< 24 hours -> Daily batch is sufficient
What is the data volume per batch?
< 100K rows -> Simple scripts, cron
100K-10M rows -> Airflow + pandas/SQL
10M rows -> Spark/distributed processing
# Pipeline configuration (environment-driven)
pipeline:
name: customer_360_daily
schedule: "0 6 * * *" # 6 AM UTC daily
timeout_minutes: 120
retries: 3
retry_delay_minutes: 15
alerts:
on_failure: ["data-eng@company.com"]
on_sla_miss: ["data-eng@company.com", "analytics@company.com"]
sla_minutes: 90
stages:
- name: extract_crm
# ... (condensed) ...
min: 1000
max: 10000000
- type: freshness
column: updated_at
max_age_hours: 48Key metrics for ETL pipelines:
Throughput: Records processed per second
Latency: Time from source change to target availability
Error rate: Percentage of records that failed transformation
Data freshness: Age of the most recent record in the target
Pipeline duration: Total wall-clock time per run
Resource utilization: CPU, memory, I/O during pipeline execution
# Pipeline metrics emission
class PipelineMetrics:
def __init__(self, pipeline_name, metrics_backend):
self.pipeline = pipeline_name
self.backend = metrics_backend
self.start_time = None
def start_run(self):
self.start_time = time.time()
self.backend.gauge(f'pipeline.{self.pipeline}.running', 1)
def end_run(self, status, records_processed, records_failed):
duration = time.time() - self.start_time
self.backend.gauge(f'pipeline.{self.pipeline}.running', 0)
self.backend.timer(f'pipeline.{self.pipeline}.duration', duration)
self.backend.counter(f'pipeline.{self.pipeline}.records.processed', records_processed)
self.backend.counter(f'pipeline.{self.pipeline}.records.failed', records_failed)
self.backend.counter(f'pipeline.{self.pipeline}.runs.{status}', 1)
if records_processed > 0:
error_rate = records_failed / (records_processed + records_failed) * 100
self.backend.gauge(f'pipeline.{self.pipeline}.error_rate', error_rate)| Factor | ETL | ELT |
|---|---|---|
| Compute cost | Source/pipeline server | Target warehouse |
| Data volume | Better for filtering before load | Better when warehouse is powerful |
| Complexity | Transformations in code | Transformations in SQL |
| Schema flexibility | More control | Relies on warehouse capabilities |
| Latency | May add latency for transforms | Load first, transform on read |
| Best for | Legacy systems, complex logic | Cloud warehouses (Snowflake, BigQuery) |
Use this skill when:
Designing or implementing etl architect solutions
Reviewing or improving existing etl architect approaches
Making architectural or implementation decisions about etl architect
Learning etl architect patterns and best practices
Troubleshooting etl architect-related issues
Do NOT use this skill when:
The question is about a fundamentally different technology domain
A more specific sibling skill covers the exact topic needed
The user needs a complete hands-on tutorial rather than expert guidance
# Etl Architect Analysis
## Context Assessment
[Situation summary and constraints]
## Recommended Approach
[Primary recommendation with rationale]
## Implementation Steps
1. [Step with specific details]
2. [Step with specific details]
3. [Step with specific details]
## Trade-offs and Considerations
- [Key trade-off 1]
- [Key trade-off 2]
## Next Steps
- [Immediate action item]
- [Follow-up action item]Input: "Help me implement etl architect for a medium-scale production application"
Output: A structured analysis covering current state assessment, recommended etl architect approach with specific patterns, implementation roadmap with milestones, and risk mitigation strategies tailored to the application scale and constraints.
Legacy system integration: When etl architect must coexist with legacy approaches, provide a gradual migration path rather than a complete rewrite
Scale mismatch: When the solution complexity exceeds the project scale, recommend a simpler approach and note when to revisit
Team skill gaps: When the team lacks experience with the recommended approach, include learning resources and simpler alternatives
Conflicting requirements: When constraints conflict (e.g., performance vs. maintainability), explicitly state the trade-off and recommend based on stated priorities
SKILL.md
430 lines · 1,676 words
The text of SKILL.md is shown as it downloads, under the Apache-2.0 license, by Ferrox Labs. The license is the one in the file's own frontmatter.
Your new bot
ETL Architect bot
Waiting in your sidebar once you say go.
No account needed · Runs on the AI plan you already pay for