data-pipeline

Data Pipeline

建立資料管線與ETL工作流程，用於資料整合、轉換與分析自動化。基於n8n的資料工作流程範本。

概述

此技能涵蓋：

• 從多個來源提取資料
• 轉換與清理
• 載入至目標位置
• 排程與監控
• 錯誤處理與警示

---

ETL 模式

基本 ETL 流程

┌─────────────┐    ┌─────────────┐    ┌─────────────┐
│   提取      │───▶│   轉換      │───▶│   載入      │
│             │    │             │    │             │
│ • API       │    │ • 清理      │    │ • 資料庫    │
│ • 資料庫    │    │ • 對應      │    │ • 資料倉儲  │
│ • 檔案      │    │ • 彙總      │    │ • 檔案      │
│ • Webhook   │    │ • 豐富化    │    │ • API       │
└─────────────┘    └─────────────┘    └─────────────┘

n8n ETL 工作流程

workflow: "每日銷售ETL"
schedule: "每日凌晨2點"

nodes:
  # 提取
  - name: "從Shopify提取"
    type: shopify
    action: get_orders
    filter: created_at >= yesterday
    
  - name: "從Stripe提取"
    type: stripe
    action: get_payments
    filter: created >= yesterday
    
  # 轉換
  - name: "合併資料"
    type: merge
    mode: combine_by_key
    key: order_id
    
  - name: "轉換"
    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
      }));
      
  # 載入
  - name: "載入至BigQuery"
    type: google_bigquery
    action: insert_rows
    table: sales_daily
    
  - name: "更新Google Sheets"
    type: google_sheets
    action: append_rows
    spreadsheet: "每日銷售報表"

---

資料來源

常見提取器

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:
  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

程式碼轉換範例

// 清理並標準化資料
function transform(items) {
  return items.map(item => ({
    // 清理字串
    name: item.name?.trim().toLowerCase(),
    
    // 解析日期
    date: new Date(item.created_at).toISOString().split('T')[0],
    
    // 轉換型別
    amount: parseFloat(item.amount) || 0,
    
    // 對應值
    status: statusMap[item.status_code] || 'unknown',
    
    // 計算欄位
    total: item.quantity * item.unit_price,
    
    // 過濾巢狀
    tags: item.tags?.filter(t => t.active).map(t => t.name),
    
    // 預設值
    source: item.source || 'direct'
  }));
}

// 彙總資料
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);
}

---

資料目的地

常見載入器

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:
  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:
  metrics:
    - rows_processed
    - execution_time
    - error_count
    - data_freshness
    
  alerts:
    pipeline_failed:
      channels: [slack, pagerduty]
      template: |
        🚨 *管線失敗*
        
        管線：{pipeline_name}
        階段：{failed_stage}
        錯誤：{error_message}
        
        [檢視日誌]({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:
  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

---

輸出範例

請求：「建立一個每日銷售資料管線」

輸出：

# 每日銷售資料管線

## 管線概覽

Shopify + Stripe → 轉換 → BigQuery + Sheets

## 排程
- 執行時間：每日凌晨2點
- 時區：UTC
- 重試次數：3次

## 提取

### Shopify訂單
```yaml
source: shopify
filter: created_at >= yesterday
fields: [id, email, total_price, line_items, created_at]

Stripe付款

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

轉換

// 合併並清理資料
{
  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
}

載入

BigQuery

• 資料表：analytics.sales_daily
• 模式：附加

Google Sheets

• 試算表：「每日銷售儀表板」
• 分頁：「原始資料」

品質檢查

• [ ] 資料列數 > 0
• [ ] 無空值order_id
• [ ] 營收總額與Stripe相符

警示

• Slack：#data-alerts
• 失敗時：@data-team

---

*Data Pipeline Skill - Part of Claude Office Skills*