构建 CSV 到 SQLite 的 ETL 管道

项目目标

本项目用三个小函数实现可用的 ETL，并让“拒绝路径”与正常路径同等重要。行从 CSV 流经模式驱动的转换进入 SQLite，任何校验失败都会被连同原因收集而非中断载入，摘要则如实给出各阶段计数。

核心特性

• 从多个 CSV 抽取带表头的原始行
• 按声明模式校验并类型转换每行
• 拒绝非法行并给出原因且继续处理其余
• 单个事务内批量载入 SQLite

架构与分层

抽取阶段把 CSV 读为原始行；转换阶段应用模式并把行分成通过与被拒两桶并附原因；载入阶段用预处理语句在单事务内写入；报告阶段汇总各阶段。除载入外每个阶段都是纯函数。

实施阶段

六个阶段为：抽取、转换、载入、报告、管道接线、CLI 与测试。中途失败回滚是事务存在的意义所在。

测试与验收

用 vitest 运行。转换测试覆盖缺失必填列、整数转换失败、可选空值与非空空白；载入测试把通过行写入内存库并断言行数，且断言错误时事务回滚；管道测试端到端跑小 CSV 并断言摘要计数。

延伸方向

JSON 或接口等新来源是新的抽取器，新目标表是模式声明，干跑模式是只做转换不载入。阶段间以普通数组通信，因此每个都可独立测试。

本译文由 AI 依据英文原文完整重写，覆盖目标、特性、架构、分阶段实现、测试与扩展方向，供中文读者对照英文原稿阅读。

代码示例

export function transform(rows, schema) {
  const accepted = [];
  const rejected = [];
  rows.forEach((row, index) => {
    const out = {};
    for (const col of schema) {
      const raw = row[col.name];
      if (raw == null || raw === '') {
        if (col.required) return rejected.push({ index, reason: 'missing ' + col.name });
        out[col.name] = null;
        continue;
      }
      const value = col.type === 'integer' ? Number.parseInt(raw, 10) : raw;
      if (col.type === 'integer' && Number.isNaN(value)) {
        return rejected.push({ index, reason: 'bad integer in ' + col.name });
      }
      out[col.name] = value;
    }
    accepted.push(out);
  });
  return { accepted, rejected };
}

mkdir etl-pipeline && cd etl-pipeline
npm init -y
npm install -D vitest@^2.1.0

{
  "name": "etl-pipeline",
  "type": "module",
  "scripts": {
    "etl": "node src/cli.js",
    "test": "vitest run"
  }
}

import { readFile } from 'node:fs/promises';

export function parseCsv(text) {
  const rows = text.trim().split(/\r?\n/).map((line) => line.split(','));
  const header = rows.shift();
  return rows.filter((r) => r.length > 1).map((cells) => {
    const row = {};
    header.forEach((h, i) => { row[h.trim()] = (cells[i] ?? '').trim(); });
    return row;
  });
}

export async function extract(paths) {
  const rows = [];
  for (const path of paths) rows.push(...parseCsv(await readFile(path, 'utf8')));
  return rows;
}

export function transform(rows, schema) {
  const accepted = [];
  const rejected = [];
  rows.forEach((row, index) => {
    const out = {};
    let reason = null;
    for (const col of schema) {
      const raw = row[col.name];
      if (!raw) {
        if (col.required) { reason = 'missing ' + col.name; break; }
        out[col.name] = null;
        continue;
      }
      if (col.type === 'integer') {
        const n = Number.parseInt(raw, 10);
        if (Number.isNaN(n)) { reason = 'bad integer in ' + col.name; break; }
        out[col.name] = n;
      } else out[col.name] = raw;
    }
    if (reason) rejected.push({ index, reason });
    else accepted.push(out);
  });
  return { accepted, rejected };
}

export function load(db, table, columns, rows) {
  const placeholders = columns.map(() => '?').join(', ');
  const stmt = db.prepare('INSERT INTO ' + table + ' (' + columns.join(', ') + ') VALUES (' + placeholders + ')');
  let inserted = 0;
  db.exec('BEGIN');
  try {
    for (const row of rows) {
      stmt.run(...columns.map((c) => row[c] ?? null));
      inserted++;
    }
    db.exec('COMMIT');
    return inserted;
  } catch (err) {
    db.exec('ROLLBACK');
    throw err;
  }
}

export function summarize({ read, accepted, rejected, loaded }) {
  return [
    'read:     ' + read,
    'accepted: ' + accepted,
    'rejected: ' + rejected.length,
    'loaded:   ' + loaded,
  ].join('\n');
}

export function rejectionSample(rejected, limit = 5) {
  return rejected.slice(0, limit).map((r) => 'row ' + r.index + ': ' + r.reason);
}

import { extract } from './extract.js';
import { transform } from './transform.js';
import { load } from './load.js';

export async function runPipeline({ db, paths, schema, table }) {
  const raw = await extract(paths);
  const { accepted, rejected } = transform(raw, schema);
  const columns = schema.map((c) => c.name);
  const types = schema.map((c) => c.name + ' ' + (c.type === 'integer' ? 'INTEGER' : 'TEXT')).join(', ');
  db.exec('CREATE TABLE IF NOT EXISTS ' + table + ' (' + types + ')');
  const loaded = load(db, table, columns, accepted);
  return { read: raw.length, accepted: accepted.length, rejected, loaded };
}

import { readFile } from 'node:fs/promises';
import { DatabaseSync } from 'node:sqlite';
import { runPipeline } from './pipeline.js';
import { summarize } from './report.js';

const [csvArgs, schemaPath, dbPath] = [process.argv.slice(2, -2), process.argv.at(-2), process.argv.at(-1)];
const schema = JSON.parse(await readFile(schemaPath, 'utf8'));
const db = new DatabaseSync(dbPath ?? ':memory:');
const result = await runPipeline({ db, paths: csvArgs, schema, table: 'records' });
console.log(summarize(result));