ETL i Pipelines de Dades amb n8n

Fonaments d'ETL

Extract, Transform, Load:

EXTRACT (E):
  ├── APIs (REST, GraphQL)
  ├── Bases de dades (PostgreSQL, MySQL, MongoDB)
  ├── Fitxers (CSV, JSON, XML, Excel)
  ├── Web scraping
  └── Streaming (Webhooks, Kafka)

TRANSFORM (T):
  ├── Neteja (nulls, duplicats, outliers)
  ├── Validació (formats, rangs, tipus)
  ├── Enriquiment (lookup, API calls)
  ├── Agregació (group by, sum, avg)
  └── Normalització (formats consistents)

LOAD (L):
  ├── Data Warehouses (BigQuery, Snowflake)
  ├── Bases de dades (PostgreSQL, MongoDB)
  ├── Data Lakes (S3, GCS)
  ├── APIs (enviar dades processades)
  └── Cache (Redis, Memcached)

N8N vs eines tradicionals:

Apache Airflow:
  ✅ Millor per a pipelines molt complexos
  ✅ Més escalable per big data massiu
  ❌ Només codi (Python DAGs)
  ❌ Configuració complexa

N8N:
  ✅ Visual + codi
  ✅ Setup ràpid (minuts vs dies)
  ✅ Ideal per SMB i mid-market
  ✅ Excel·lent per a prototipatge
  ❌ Menys adequat per a petabytes

Casos d'ús N8N en Big Data:

✅ Excel·lent per:
- Pipelines fins a 10M registres/dia
- Integracions entre 3-10 sistemes
- ETL amb transformacions mitjanes
- Prototipatge ràpid de pipelines

⚠️ Acceptable amb optimitzacions:
- 10M-50M registres/dia (amb batching)
- Transformacions pesades (Code Node + external processing)

❌ No recomanat:
- >100M registres/dia
- Streaming en temps real massiu (millor Kafka+Flink)
- Processament de fitxers >1GB in-memory

Extracció de dades

APIs REST:

Exemple complet: Extreure dades de Salesforce

[Schedule Trigger: Daily 2AM]
  ↓
[HTTP Request: Salesforce Auth]
  Method: POST
  URL: https://login.salesforce.com/services/oauth2/token
  Body:
    grant_type: password
    client_id: {{$env.SF_CLIENT_ID}}
    client_secret: {{$env.SF_CLIENT_SECRET}}
    username: {{$env.SF_USERNAME}}
    password: {{$env.SF_PASSWORD}}
  ↓
[Set: Extract Token]
  access_token: {{$json.access_token}}
  ↓
[HTTP Request: Get Accounts]
  URL: https://{{$env.SF_INSTANCE}}.salesforce.com/services/data/v58.0/query
  Query Params:
    q: SELECT Id, Name, Industry, AnnualRevenue FROM Account WHERE LastModifiedDate > {{$json.last_sync_date}}
  Headers:
    Authorization: Bearer {{$node["Extract Token"].json.access_token}}

GraphQL APIs:

[HTTP Request]
  Method: POST
  URL: https://api.github.com/graphql
  Headers:
    Authorization: Bearer {{$credentials.github_token}}
  Body:
    {
      "query": "query { 
        repository(owner: \"n8n-io\", name: \"n8n\") {
          issues(first: 100, states: OPEN) {
            edges {
              node {
                title
                createdAt
                author { login }
                labels(first: 10) {
                  edges { node { name } }
                }
              }
            }
          }
        }
      }"
    }

Bases de dades:

Estratègia Incremental Load:

[PostgreSQL: Get Last Sync Time]
  Query: SELECT MAX(sync_timestamp) as last_sync FROM sync_log
  ↓
[PostgreSQL: Get New/Modified Records]
  Query:
    SELECT * FROM customers
    WHERE updated_at > '{{$node["Get Last Sync Time"].json.last_sync}}'
    ORDER BY updated_at
    LIMIT 10000
  ↓
[Transform]
  ↓
[Load to DWH]
  ↓
[PostgreSQL: Update Sync Log]
  Query: INSERT INTO sync_log (sync_timestamp, records_processed) 
         VALUES ('{{$now}}', {{$json.count}})

Fitxers CSV/Excel:

[Read Binary File]
  File Path: /data/uploads/sales_{{$json.date}}.csv
  ↓
[Spreadsheet File]
  Operation: Read from File
  File Format: CSV
  Options:
    - Header Row: Yes
    - Delimiter: ,
    - Encoding: UTF-8
  ↓
[Function: Parse and Validate]
  // Validar format dates, números, etc.

Web Scraping:

[HTTP Request]
  URL: https://example.com/products
  ↓
[HTML Extract]
  Extraction Values:
    - Selector: .product-title
      Attribute: text
      Key: title
    - Selector: .product-price
      Attribute: text
      Key: price
    - Selector: .product-link
      Attribute: href
      Key: url
  ↓
[Function: Clean Data]
  price: parseFloat(price.replace('€', ''))

Transformació de dades

Neteja de dades:

// Function Node: Data Cleaning
const items = $input.all();

return items
  .map(item => {
    const data = item.json;

    return {
      json: {
        // Remove nulls and empty strings
        id: data.id || null,
        name: data.name?.trim() || 'Unknown',
        email: data.email?.toLowerCase().trim() || null,

        // Normalize phone numbers
        phone: data.phone?.replace(/[^0-9+]/g, '') || null,

        // Fix dates
        created_at: data.created_at ? 
          new Date(data.created_at).toISOString() : 
          null,

        // Remove outliers (z-score method)
        amount: Math.abs(data.amount) < 1000000 ? data.amount : null,

        // Standardize categories
        category: data.category?.toLowerCase()
          .replace(/[^a-z0-9]/g, '_') || 'uncategorized'
      }
    };
  })
  .filter(item => {
    // Remove invalid records
    return item.json.id && item.json.email;
  });

Normalització:

// Normalize addresses
function normalizeAddress(addr) {
  return {
    street: addr.street?.trim(),
    city: addr.city?.trim().toUpperCase(),
    postal_code: addr.postal_code?.replace(/\s/g, ''),
    country: addr.country?.trim().toUpperCase() || 'ES'
  };
}

// Normalize currencies
function normalizeCurrency(amount, from, to = 'EUR') {
  const rates = {
    'USD': 0.92,
    'GBP': 1.17,
    'EUR': 1.00
  };

  return amount * rates[from] / rates[to];
}

Validació:

// Validation rules
const validationRules = {
  email: /^[^\s@]+@[^\s@]+\.[^\s@]+$/,
  phone: /^\+?[0-9]{9,15}$/,
  postalCode: /^[0-9]{5}$/,
  iban: /^ES[0-9]{22}$/
};

function validate(data, rules) {
  const errors = [];

  for (const [field, regex] of Object.entries(rules)) {
    if (data[field] && !regex.test(data[field])) {
      errors.push(`Invalid ${field}: ${data[field]}`);
    }
  }

  return {
    isValid: errors.length === 0,
    errors: errors
  };
}

// Apply validation
const items = $input.all();
return items.map(item => {
  const validation = validate(item.json, validationRules);

  return {
    json: {
      ...item.json,
      _validation: validation,
      _is_valid: validation.isValid
    }
  };
});

Enriquiment:

[Base Data]
  ↓
[HTTP Request: Enrich with Geolocation]
  URL: https://api.ipgeolocation.io/ipgeo
  Params:
    apiKey: {{$env.GEO_API_KEY}}
    ip: {{$json.ip_address}}
  ↓
[PostgreSQL: Enrich with Customer Tier]
  Query:
    SELECT tier, discount_rate 
    FROM customer_tiers 
    WHERE customer_id = {{$json.customer_id}}
  ↓
[Merge: Combine All Data]
  Mode: Keep Key Matches
  Key: customer_id
  ↓
[Enriched Data]

Agregacions:

// Group by and aggregate
const items = $input.all();
const grouped = {};

items.forEach(item => {
  const category = item.json.category;

  if (!grouped[category]) {
    grouped[category] = {
      category: category,
      count: 0,
      total_amount: 0,
      items: []
    };
  }

  grouped[category].count++;
  grouped[category].total_amount += item.json.amount;
  grouped[category].items.push(item.json);
});

// Calculate aggregates
return Object.values(grouped).map(group => ({
  json: {
    category: group.category,
    total_count: group.count,
    total_amount: group.total_amount,
    avg_amount: group.total_amount / group.count,
    min_amount: Math.min(...group.items.map(i => i.amount)),
    max_amount: Math.max(...group.items.map(i => i.amount))
  }
}));

Càrrega de dades

PostgreSQL/MySQL:

[Transform] (10,000 records)
  ↓
[Split In Batches: 500]
  ↓
[PostgreSQL: Bulk Insert]
  Operation: Insert
  Table: staging.sales_data
  Columns: Auto-map
  Options:
    - On Conflict: Update
    - Returning: id, created_at
  ↓
[Loop until all loaded]

BigQuery (Data Warehouse):

[Prepare Data]
  ↓
[HTTP Request: BigQuery API]
  Method: POST
  URL: https://bigquery.googleapis.com/bigquery/v2/projects/{{$env.GCP_PROJECT}}/datasets/{{$env.DATASET}}/tables/{{$env.TABLE}}/insertAll
  Headers:
    Authorization: Bearer {{$credentials.gcp_token}}
  Body:
    {
      "rows": [
        {{$json.rows.map(r => ({json: r}))}}
      ],
      "skipInvalidRows": false,
      "ignoreUnknownValues": false
    }

S3/Cloud Storage:

[Transform to CSV]
  ↓
[Convert to Binary]
  ↓
[S3: Upload File]
  Bucket: company-data-lake
  Path: /raw/sales/{{$now.toFormat('yyyy-MM-dd')}}/sales_data.csv
  ACL: private
  Metadata:
    - source: n8n-workflow
    - execution_id: {{$execution.id}}

Webhooks (notificar sistemes externs):

[ETL Complete]
  ↓
[HTTP Request: Notify Data Warehouse]
  Method: POST
  URL: {{$env.DWH_WEBHOOK_URL}}
  Body:
    {
      "event": "etl_complete",
      "workflow": "{{$workflow.name}}",
      "records_processed": {{$json.count}},
      "execution_time_ms": {{$json.duration}},
      "timestamp": "{{$now}}"
    }

Patrons de disseny en pipelines

Incremental Loading:

[Get Last Watermark]
  Query: SELECT MAX(updated_at) as watermark FROM staging.sync_metadata
  ↓
[Extract New/Modified]
  Query: SELECT * FROM source 
         WHERE updated_at > '{{$json.watermark}}'
  ↓
[Transform]
  ↓
[Load]
  ↓
[Update Watermark]
  Query: UPDATE staging.sync_metadata 
         SET updated_at = '{{$now}}'

Change Data Capture (CDC):

[PostgreSQL: Read WAL/Replication Slot]
  or
[Webhook: Receive CDC Events]
  ↓
[Parse Change Event]
  event_type: INSERT | UPDATE | DELETE
  table: customers
  data: {...}
  ↓
[Switch by Event Type]
  ├─ INSERT → [Load New Record]
  ├─ UPDATE → [Update Existing]
  └─ DELETE → [Soft Delete]

Checkpointing:

// Checkpoint pattern for resilience
const checkpoint = {
  batch_id: Date.now(),
  last_processed_id: 0,
  total_processed: 0
};

// Load checkpoint if exists
try {
  const lastCheckpoint = await loadCheckpoint();
  if (lastCheckpoint) {
    checkpoint.last_processed_id = lastCheckpoint.last_processed_id;
  }
} catch (e) {}

// Process with checkpoints
while (true) {
  const batch = await fetchBatch(checkpoint.last_processed_id);
  if (batch.length === 0) break;

  await processBatch(batch);

  checkpoint.last_processed_id = batch[batch.length - 1].id;
  checkpoint.total_processed += batch.length;

  // Save checkpoint every 1000 records
  if (checkpoint.total_processed % 1000 === 0) {
    await saveCheckpoint(checkpoint);
  }
}

Idempotència:

-- Pattern 1: UPSERT with ON CONFLICT
INSERT INTO customers (id, name, email, updated_at)
VALUES ($1, $2, $3, $4)
ON CONFLICT (id) 
DO UPDATE SET 
  name = EXCLUDED.name,
  email = EXCLUDED.email,
  updated_at = EXCLUDED.updated_at;

-- Pattern 2: Check before insert
BEGIN;
DELETE FROM staging.temp_data WHERE batch_id = '{{$json.batch_id}}';
INSERT INTO staging.temp_data ...;
COMMIT;

-- Pattern 3: Idempotent processing key
INSERT INTO processed_events (event_id, processed_at)
VALUES ('{{$json.event_id}}', NOW())
ON CONFLICT (event_id) DO NOTHING
RETURNING event_id;
-- If returns null, event was already processed

Gestió de grans volums:

Strategy 1: Pagination + Batching
[Get Total Count]
  ↓
[Calculate Pages] (100,000 records / 1,000 per page = 100 pages)
  ↓
[Loop: Page 1 to 100]
  ↓
  [Fetch Page]
    ↓
  [Split In Batches: 100]
    ↓
  [Process Batch]
    ↓
  [Load Batch]
  ↓
[Next Page]

Strategy 2: Parallel Processing
[Source Data]
  ↓
[Split by Key Range]
  ├─ Range 1: IDs 1-10000 → Worker 1
  ├─ Range 2: IDs 10001-20000 → Worker 2
  └─ Range 3: IDs 20001-30000 → Worker 3
  ↓
[Merge Results]

Strategy 3: Streaming
[Webhook Trigger: Continuous]
  ↓
[Buffer: Collect for 60s or 1000 items]
  ↓
[Process Buffer]
  ↓
[Load Batch]
  ↓
[Repeat]