メニュー
Design ETL/ELT pipelines with proper orchestration, error handling, and monitoring. Use when building data pipelines, designing data workflows, or implementing data transformations.
Codex または Claude でインストール この Prompt をコピーして Codex、Claude、または他のアシスタントに貼り付けると、Skill ページを確認してインストールできます。
SOC 職業分類に基づく
Creates and validates Azure Resource Manager (ARM) templates for infrastructure deployment. Use when creating ARM templates, deploying Azure infrastructure as code, or validating Azure templates.
Analyzes Azure costs and provides optimization recommendations including reserved instances, rightsizing, and unused resources. Use when optimizing Azure spending or analyzing Azure costs.
Creates dbt models with proper layering (staging, marts), incremental strategies, and documentation. Use when creating dbt models, organizing data transformations, or implementing incremental models.
Generates dbt tests including schema tests, data quality tests, and freshness checks. Use when adding tests to dbt models or implementing data quality validation.
Configures Google Cloud Build pipelines with caching, parallel builds, and optimization. Use when setting up Cloud Build, optimizing build performance, or configuring CI/CD pipelines.
Analyzes GCP costs and provides optimization recommendations including committed use discounts, rightsizing, and unused resources. Use when optimizing GCP spending or analyzing GCP costs.
| name | etl-designer |
| description | Design ETL/ELT pipelines with proper orchestration, error handling, and monitoring. Use when building data pipelines, designing data workflows, or implementing data transformations. |
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': ['alerts@company.com']
}
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