Data Engineering

knowledge skill

The agent constructs ETL pipelines to move data from source systems to a target warehouse. It produces executable SQL scripts or Python code that performs schema mapping, data cleaning, and loading operations.

Worked examples

• {"input": {"source_schema": "users_csv: {id: int, full_name: string, created_at: string}", "target_destination": "dw.dim_users", "transformation_rules": "Split full_name into first_name and last_name; cast created_at to timestamp"}, "output": {"pipeline_code": "INSERT INTO dw.dim_users (user_id, first_name, last_name, created_date) SELECT id, split_part(full_name, ' ', 1), split_part(full_name, ' ', 2), created_at::timestamp FROM raw_users;", "schema_mapping": {"id": "user_id", "full_name": ["first_name", "last_name"], "created_at": "created_date"}, "execution_plan": "1. Load CSV to staging; 2. Execute transformation SQL; 3. Update dim_users"}}
• {"input": {"source_schema": "api_logs: {timestamp: string, level: string, msg: string}", "target_destination": "logs_parquet_s3", "transformation_rules": "Filter for level='ERROR'; aggregate counts by hour"}, "output": {"pipeline_code": "df = spark.read.json('api_logs').filter('level == \"ERROR\"').groupBy(window('timestamp', '1 hour')).count().write.parquet('s3://logs_parquet_s3')", "schema_mapping": {"timestamp": "window_start", "level": "filtered_error", "msg": "dropped"}, "execution_plan": "1. Read JSON stream; 2. Filter by level; 3. Window aggregation; 4. Write Parquet"}}

Input

• source_schema: JSON/DDL of origin data
• target_destination: Warehouse table name or URI
• transformation_rules: List of mapping and filtering logic

Output

• pipeline_code: SQL or Python script for data movement
• schema_mapping: Map of source columns to target columns
• execution_plan: Sequence of load operations

Details

• Skill type: knowledge skill
• Safety level: safe_public_research
• Version: 1.0.0