CSV 流式解析与增量持久化实践

概述CSV 常用于批量数据交换。本文以行拆分与简单分隔解析为基础，演示增量持久化路径。解析管道（基础版）function csvSplitter(headers) { let buf = ''; return new TransformStream({ transform(chunk, controller) { buf += chunk; const lines = buf.split('\n'); buf = lines.pop(); for (const line of lines) { const cols = line.split(',').map(s => s.trim()); const obj = Object.fromEntries(headers.map((h, i) => [h, cols[i] || ''])); controller.enqueue(obj); } }, flush(controller) { if (buf) { const cols = buf.split(',').map(s => s.trim()); const obj = Object.fromEntries(headers.map((h, i) => [h, cols[i] || ''])); controller.enqueue(obj); } } }); } 增量写入 IndexedDBfunction openDB(name) { return new Promise((resolve, reject) => { const r = indexedDB.open(name, 1); r.onupgradeneeded = () => { const db = r.result; if (!db.objectStoreNames.contains('rows')) db.createObjectStore('rows', { keyPath: 'id' }); }; r.onsuccess = () => resolve(r.result); r.onerror = () => reject(r.error); }); } async function importCSV(url, headers) { const res = await fetch(url); const rs = res.body.pipeThrough(new TextDecoderStream()).pipeThrough(csvSplitter(headers)); const reader = rs.getReader(); const db = await openDB('csvdb'); let id = 1; while (true) { const { value, done } = await reader.read(); if (done) break; const tx = db.transaction('rows', 'readwrite'); tx.objectStore('rows').put({ id: id++, ...value }); await new Promise((resolve, reject) => { tx.oncomplete = resolve; tx.onerror = () => reject(tx.error); }); } db.close(); } 注意事项基础版不处理引号与转义，复杂 CSV 请引入专用解析器或增强管道。导入过程建议增加校验与失败重试。