data-pipeline（数据管道）

Data Pipeline

Build data pipelines and ETL workflows for data integration, transformation, and analytics automation. Based on n8n's data workflow templates.

Overview

This skill covers:

• Data extraction from multiple sources
• Transformation and cleaning
• Loading to destinations
• Scheduling and monitoring
• Error handling and alerts

ETL Patterns

Basic ETL Flow

┌─────────────┐    ┌─────────────┐    ┌─────────────┐
│   EXTRACT   │───▶│  TRANSFORM  │───▶│    LOAD     │
│             │    │             │    │             │
│ • APIs      │    │ • Clean     │    │ • Database  │
│ • Databases │    │ • Map       │    │ • Warehouse │
│ • Files     │    │ • Aggregate │    │ • Files     │
│ • Webhooks  │    │ • Enrich    │    │ • APIs      │
└─────────────┘    └─────────────┘    └─────────────┘

n8n ETL Workflow

workflow: "Daily Sales ETL"
schedule: "2am daily"

nodes:
  # EXTRACT
  - name: "Extract from Shopify"
    type: shopify
    action: get_orders
    filter: created_at >= yesterday

  - name: "Extract from Stripe"
    type: stripe
    action: get_payments
    filter: created >= yesterday

  # TRANSFORM
  - name: "Merge Data"
    type: merge
    mode: combine_by_key
    key: order_id

  - name: "Transform"
    type: code
    code: |
      return items.map(item => ({
        date: item.created_at.split('T')[0],
        order_id: item.id,
        customer_email: item.email,
        total: parseFloat(item.total_price),
        currency: item.currency,
        items: item.line_items.length,
        source: item.source_name,
        payment_status: item.payment.status
      }));

  # LOAD
  - name: "Load to BigQuery"
    type: google_bigquery
    action: insert_rows
    table: sales_daily

  - name: "Update Google Sheets"
    type: google_sheets
    action: append_rows
    spreadsheet: "Daily Sales Report"

Data Sources

Common Extractors

extractors:
  databases:
    - postgresql:
        connection: connection_string
        query: "SELECT * FROM orders WHERE date >= $1"

    - mysql:
        connection: connection_string
        query: custom_sql

    - mongodb:
        connection: connection_string
        collection: orders
        filter: {date: {$gte: yesterday}}

  apis:
    - rest_api:
        url: "https://api.example.com/data"
        method: GET
        headers: {Authorization: "Bearer {token}"}
        pagination: handle_automatically

    - graphql:
        url: "https://api.example.com/graphql"
        query: graphql_query

  files:
    - csv:
        source: sftp/s3/google_drive
        delimiter: ","
        encoding: utf-8

    - excel:
        source: file_path
        sheet: "Sheet1"

    - json:
        source: api/file
        path: "data.items"

  saas:
    - salesforce: get_objects
    - hubspot: get_contacts/deals
    - stripe: get_charges
    - shopify: get_orders

Transformations

Common Transformations

transformations:
  cleaning:
    - remove_nulls: drop_or_fill
    - trim_whitespace: all_string_fields
    - deduplicate: by_key
    - validate: against_schema

  mapping:
    - rename_fields: {old_name: new_name}
    - convert_types: {date_string: date}
    - map_values: {status_code: status_name}

  aggregation:
    - group_by: [date, category]
    - sum: [revenue, quantity]
    - count: orders
    - average: order_value

  enrichment:
    - lookup: from_reference_table
    - geocode: from_address
    - calculate: derived_fields

  filtering:
    - where: condition
    - limit: n_rows
    - sample: percentage

Code Transform Examples

// Clean and normalize data
function transform(items) {
  return items.map(item => ({
    // Clean strings
    name: item.name?.trim().toLowerCase(),

    // Parse dates
    date: new Date(item.created_at).toISOString().split('T')[0],

    // Convert types
    amount: parseFloat(item.amount) || 0,

    // Map values
    status: statusMap[item.status_code] || 'unknown',

    // Calculate fields
    total: item.quantity * item.unit_price,

    // Filter nested
    tags: item.tags?.filter(t => t.active).map(t => t.name),

    // Default values
    source: item.source || 'direct'
  }));
}

// Aggregate data
function aggregate(items) {
  const grouped = {};

  items.forEach(item => {
    const key = `${item.date}_${item.category}`;
    if (!grouped[key]) {
      grouped[key] = {
        date: item.date,
        category: item.category,
        total_revenue: 0,
        order_count: 0
      };
    }
    grouped[key].total_revenue += item.amount;
    grouped[key].order_count += 1;
  });

  return Object.values(grouped);
}

Data Destinations

Common Loaders

loaders:
  data_warehouses:
    - bigquery:
        project: project_id
        dataset: analytics
        table: sales
        write_mode: append/truncate

    - snowflake:
        account: account_id
        warehouse: compute_wh
        database: analytics
        schema: public

    - redshift:
        cluster: cluster_id
        database: analytics

  databases:
    - postgresql:
        upsert: on_conflict_update

    - mysql:
        batch_insert: 1000_rows

  files:
    - s3:
        bucket: data-lake
        path: /processed/{date}/
        format: parquet

    - google_cloud_storage:
        bucket: data-bucket

  spreadsheets:
    - google_sheets:
        mode: append/overwrite

    - airtable:
        base: base_id
        table: table_name

  apis:
    - webhook:
        url: destination_url
        batch_size: 100

Scheduling & Monitoring

Pipeline Scheduling

scheduling:
  patterns:
    hourly:
      cron: "0 * * * *"
      use_for: real_time_dashboards

    daily:
      cron: "0 2 * * *"
      use_for: daily_reports

    weekly:
      cron: "0 3 * * 1"
      use_for: weekly_summaries

    on_demand:
      trigger: webhook/manual
      use_for: ad_hoc_analysis

  dependencies:
    - pipeline_a: must_complete_before pipeline_b
    - wait_for: all_extracts_complete

  retries:
    max_attempts: 3
    delay: exponential_backoff
    alert_on: final_failure

Monitoring & Alerts

monitoring:
  metrics:
    - rows_processed
    - execution_time
    - error_count
    - data_freshness

  alerts:
    pipeline_failed:
      channels: [slack, pagerduty]
      template: |
        🚨 *Pipeline Failed*

        Pipeline: {pipeline_name}
        Stage: {failed_stage}
        Error: {error_message}

        [View Logs]({logs_url})

    data_quality:
      trigger: anomaly_detected
      conditions:
        - row_count: differs_by > 50%
        - null_rate: exceeds_threshold
        - schema: changed_unexpectedly

    stale_data:
      trigger: last_update > threshold
      threshold: 2_hours

Data Quality

Quality Checks

data_quality:
  schema_validation:
    - required_fields: [id, date, amount]
    - field_types:
        id: integer
        date: date
        amount: number
    - allowed_values:
        status: [active, pending, closed]

  statistical_checks:
    - null_rate: < 5%
    - duplicate_rate: < 1%
    - value_range:
        amount: [0, 1000000]

  business_rules:
    - total_equals_sum_of_line_items
    - dates_are_not_in_future
    - email_format_valid

  trend_analysis:
    - row_count: within_2_std_of_mean
    - total_value: within_expected_range

Output Example

Request: "Create a daily sales data pipeline"

Output:

# Daily Sales Data Pipeline

## Pipeline Overview

Shopify + Stripe → Transform → BigQuery + Sheets

## Schedule
- Runs: 2am daily
- Timezone: UTC
- Retry: 3 attempts

## Extract

### Shopify Orders

source: shopify filter: created_at >= yesterday fields: [id, email, total_price, line_items, created_at]

### Stripe Payments

source: stripe filter: created >= yesterday fields: [id, amount, status, metadata.order_id]

## Transform

// Join and clean data { date: order.created_at.split('T')[0], order_id: order.id, customer: order.email, revenue: parseFloat(order.total_price), items: order.line_items.length, payment_status: payment.status }

## Load

### BigQuery

- Table: `analytics.sales_daily`
- Mode: Append

### Google Sheets

- Sheet: "Daily Sales Dashboard"
- Tab: "Raw Data"

## Quality Checks

- Row count > 0
- No null order_ids
- Revenue sum matches Stripe

## Alerts

- Slack: #data-alerts
- On failure: @data-team

*Data Pipeline Skill - Part of Claude Office Skills*