data-pipeline · 資料處理管線引擎

可組合的資料轉換、驗證和分析管線。像搭積木一樣處理資料。

何時使用

當用戶提到：資料清洗、資料轉換、ETL、資料驗證、資料分組、資料聚合、管道處理、批次資料處理、陣列處理

快速開始

const { Pipeline, Transformers, Validators, PipelineFactory } = require('data-pipeline/src/pipeline');

// 建立一個清洗管線
const pipeline = new Pipeline();
pipeline
  .addStage('filter', Transformers.filter(x => x.age >= 18))
  .addStage('pick', Transformers.pick(['name', 'email']))
  .addStage('sort', Transformers.sort('name', 'asc'));

const result = await pipeline.run(users);

核心 API

Pipeline

const pipeline = new Pipeline({ strict: true, context: { key: 'value' } });

// 新增階段
pipeline.addStage(name, asyncFn, { retryCount: 0, retryDelay: 100, timeout: 30000 });
pipeline.addStages([{ name, fn, options }]);

// 階段管理
pipeline.insertBefore(target, name, fn, options);
pipeline.insertAfter(target, name, fn, options);
pipeline.removeStage(name);
pipeline.toggleStage(name, enabled);

// 執行
const result = await pipeline.execute(data);  // 返回 { data, metadata }
const data = await pipeline.run(data);         // 只返回資料

// 指標
const metrics = pipeline.getMetrics();
pipeline.resetMetrics();

內建轉換器

驗證器

const schema = {
  name: { required: true, type: 'string', minLength: 1 },
  age: { type: 'number', min: 0, max: 150 },
  email: { pattern: /^[^\s@]+@[^\s@]+\.[^\s@]+$/ },
  role: { enum: ['admin', 'user'] },
  password: { validate: (v) => v.length >= 8 ? true : 'Too short' }
};

const validator = Validators.schema(schema);
const result = validator(data);
// { valid: boolean, errors: [...], totalItems, validItems }

工廠函式

// ETL 管線
const etl = PipelineFactory.createETL(extract, transforms, load);

// 資料清洗管線
const cleaner = PipelineFactory.createCleaner(schema, { defaultField: 'value' });

// 資料分析管線
const analyzer = PipelineFactory.createAnalyzer('groupKey', {
  avgVal: vals => vals.reduce((a,b) => a+b, 0) / vals.length,
  maxVal: vals => Math.max(...vals)
});

使用場景

1. 資料清洗：驗證 → 去重 → 填充預設值 → 修剪字串
2. ETL 流程：提取 → 轉換（map/filter/reduce）→ 載入
3. 資料分析：分組 → 聚合 → 排序 → 分頁
4. 資料驗證：批次驗證物件陣列，返回詳細錯誤報告
5. API 資料處理：合併多個數據源 → 重新命名欄位 → 選擇輸出欄位

   7w4.net小蔥技能。

錯誤處理

try {
  const result = await pipeline.execute(data);
} catch (err) {
  if (err instanceof PipelineError) {
    console.log('Failed at:', err.failedStage);
    console.log('Partial data:', err.lastData);
    console.log('Stage results:', err.stageResults);
  }
}

效能指標

const metrics = pipeline.getMetrics();
// {
//   pipeline: { totalRuns, totalErrors, avgTime },
//   stages: [{ name, calls, errors, avgTime }, ...]
// }

質量中等偏上。文件內容非常詳細，提供了完整的資料處理流程和豐富的配置示例，但實際功能只有說明文件，沒有可執行的程式碼工具。README 說明太少，英文內容較多而中文支援不足，對中文使用者不太友好。適合作為參考模板使用，但想直接執行需要額外開發。