00_API_Call.
md 2025-12-23
00_API_Call
Hands-on Lab: Airflow API ETL with Dynamic Task
Mapping
Overview
Pada hands-on ini, kita akan membangun pipeline ETL Apache Airflow yang:
Memanggil REST API (data user: id, name, email)
Melakukan validasi payload
Load data ke staging table secara paralel menggunakan Dynamic Task Mapping
Melakukan transformasi ke final table
Menjamin idempotency dan data consistency
Pipeline ini merepresentasikan real-world ETL pattern yang umum digunakan di data engineering.
Architecture Flow
1. Fetch users dari REST API
2. Validate payload API
3. Clear staging table (idempotent)
4. Load user ke staging (parallel, mapped)
5. Transform staging → final
6. Validate hasil akhir
Prerequisites
Apache Airflow 2.6+
Postgres database
REST API endpoint /users
Basic Python & SQL knowledge
Step 1 – Menyiapkan REST API (Mock / Real)
API diharapkan mengembalikan JSON:
[
{ "id": 1, "name": "Alice", "email": "alice@[Link]" },
{ "id": 2, "name": "Bob", "email": "bob@[Link]" }
]
1/5
00_API_Call.md 2025-12-23
Endpoint:
GET /users
Step 2 – Membuat HTTP Connection di Airflow Admin
Melalui UI Airflow
1. Buka Airflow Web UI
2. Masuk menu Admin → Connections
3. Klik + Add a new record
Isikan Field
Field Value
Connection Id users_api
Connection Type HTTP
Host [Link]
Schema http
Extra {}
⚠ Jangan isi path /users di Host
Klik Save
Step 3 – Membuat Struktur Project
dags/
├── api_users_dynamic_etl.py
├── include/
│ └── sql/
│ ├── api_clear_staging.sql
│ ├── api_insert_staging.sql
│ ├── api_transform_final.sql
│ └── api_validate_final.sql
└── src/
└── etl/
└── api_validators.py
Step 4 – SQL Files
2/5
00_API_Call.md 2025-12-23
api_clear_staging.sql
DELETE FROM api_users_staging WHERE ds = %s;
api_insert_staging.sql
INSERT INTO api_users_staging (id, name, email, ds)
VALUES (%s, %s, %s, %s);
api_transform_final.sql
INSERT INTO api_users_final (id, name, email, ds)
SELECT id, UPPER(name), email, ds
FROM api_users_staging
WHERE ds = %s
ON CONFLICT (id)
DO UPDATE SET
name = [Link],
email = [Link],
ds = [Link];
api_validate_final.sql
SELECT COUNT(*) FROM api_users_final WHERE ds = %s;
Step 5 – Validator Logic
src/etl/api_validators.py
from [Link] import AirflowFailException
def validate_api_user_payload(user: dict) -> dict:
for field in ("id", "name", "email"):
if field not in user:
raise AirflowFailException(f"Missing field {field}")
return user
Step 6 – DAG Implementation (Conceptual Explanation)
3/5
00_API_Call.md 2025-12-23
Fetch Users
Menggunakan HttpHook
Mengambil data dari connection users_api
Validate Users
Validasi struktur payload API
Fail-fast jika schema tidak valid
Clear Staging (Idempotency)
Menghapus data staging berdasarkan ds
Menjamin DAG bisa di-retry tanpa duplikasi
Load User (Dynamic Task Mapping)
Setiap user → 1 task instance
Paralel load ke staging
Transform
Memindahkan data dari staging ke final
Menggunakan UPSERT (ON CONFLICT)
Validate Final
Membandingkan row count
Menjamin data completeness
Step 7 – Dynamic Task Mapping
rows = load_user.expand(user=valid_users)
Airflow akan membuat:
load_user[0]
load_user[1]
dst…
Jumlah task mengikuti jumlah data API.
Step 8 – Trigger & Observasi
1. Trigger DAG manual
2. Perhatikan:
4/5
00_API_Call.md 2025-12-23
Graph View
Parallel load
Task mapping
3. Cek tabel staging & final
Key Learning Points
✔ HTTP Hook & Airflow Connection ✔ Dynamic Task Mapping ✔ Idempotent ETL Design ✔ SQL Best
Practices (1 SQL = 1 execute) ✔ Production-grade ETL pattern
Best Practices Notes
Jangan multi-statement SQL dengan parameter
Selalu pisahkan staging & final
Gunakan ds untuk partitioning
Tambahkan unit test untuk validator
5/5