Ночной ETL без Airflow и Kubernetes

{
  "schemaVersion": 1,
  "nodes": [
    { "id": "cron", "kind": "cronTrigger", "name": "Nightly 02:00", "position": { "x": 0, "y": 0 }, "config": { "cron": "0 2 * * *", "timezone": "UTC" } },
    { "id": "extract", "kind": "lambda", "name": "Extract daily report", "position": { "x": 240, "y": 0 }, "config": { "functionId": "etl-extract", "onError": "failPipeline" } },
    { "id": "transform", "kind": "lambda", "name": "Transform records", "position": { "x": 480, "y": 0 }, "config": { "functionId": "etl-transform", "onError": "failPipeline" } },
    { "id": "load", "kind": "lambda", "name": "Load to warehouse", "position": { "x": 720, "y": 0 }, "config": { "functionId": "etl-load", "onError": "failPipeline" } }
  ],
  "edges": [
    { "id": "e1", "sourceNodeId": "cron", "targetNodeId": "extract", "sourceHandle": "default" },
    { "id": "e2", "sourceNodeId": "extract", "targetNodeId": "transform", "sourceHandle": "success" },
    { "id": "e3", "sourceNodeId": "transform", "targetNodeId": "load", "sourceHandle": "success" }
  ]
}

export async function handler(event) {
  const date = new Date().toISOString().slice(0, 10); // YYYY-MM-DD
  const records = await externalApi.fetchDailyReport(date);
  // Store to object storage — pipeline passes storage key to next step
  const key = `etl/raw/${date}.json`;
  await storage.putJson(key, records);
  return { key, count: records.length, date };
}

export async function handler(event) {
  const { key, date } = event.previousOutput ?? {};
  const raw = await storage.getJson(key);
  const transformed = raw.map((r) => ({
    id: r.external_id,
    date,
    revenue: parseFloat(r.revenue_usd),
    region: r.region?.toLowerCase(),
    updatedAt: new Date().toISOString(),
  }));
  const outKey = `etl/transformed/${date}.json`;
  await storage.putJson(outKey, transformed);
  return { outKey, count: transformed.length, date };
}

export async function handler(event) {
  const { outKey, date } = event.previousOutput ?? {};
  const rows = await storage.getJson(outKey);
  // Idempotent load — safe to retry without duplicating rows
  await db.analytics.upsertBatch(rows, { conflictKeys: ['id', 'date'] });
  return { loaded: rows.length, date };
}