Node.js中流式处理大JSON Buffer，提取geometry与properties的方案

问题场景 #

我基于Node.js的NestJS框架开发，用multer获取上传的.json（GeoJSON）文件，需要过滤内容后保存到其他位置。因为文件可能很大，所以采用流（stream）读取，同时用streamifier处理Buffer。现有代码如下：

import { Response } from 'express';
const streamifier = require('streamifier');

function uploadFile(file) {
    const json = streamifier.createReadStream(file.buffer)
    return json.pipe(process.stdout)
}

完整的NestJS接口代码：

@Post('upload')
@ApiConsumes('multipart/form-data')
@ApiBody({
    description: 'Geojson File',
    type: FileUploadDto,
})
@UseInterceptors(FileInterceptor('file'))
uploadFile(@UploadedFile() file: Express.Multer.File, @Res() res: Response) {
    const json = streamifier.createReadStream(file.buffer)
    return json.pipe(res)
}

待处理的GeoJSON示例：

{ "type": "FeatureCollection",
  "features": [
    { "type": "Feature",
      "geometry": {"type": "Point", "coordinates": [102.0, 0.5]},
      "properties": {"prop0": "value0"}
      },
    { "type": "Feature",
      "geometry": {
        "type": "LineString",
        "coordinates": [
          [102.0, 0.0], [103.0, 1.0], [104.0, 0.0], [105.0, 1.0]
          ]
        },
      "properties": {
        "prop0": "value0",
        "prop1": 0.0
        }
      },
    { "type": "Feature",
       "geometry": {
         "type": "Polygon",
         "coordinates": [
           [ [100.0, 0.0], [101.0, 0.0], [101.0, 1.0],
             [100.0, 1.0], [100.0, 0.0] ]
           ]

       },
       "properties": {
         "prop0": "value0",
         "prop1": {"this": "that"}
         }
       }
    ]
  }

需求：流式读取文件的同时，只提取每个feature里的geometry和properties字段，该怎么实现？

解决方案 #

处理大JSON流的核心是用JSON流式解析器，避免一次性加载整个文件到内存。这里用JSONStream库，它能精准定位JSON路径，提取目标字段，配合Node.js流API完成转换和输出。

1. 安装依赖 #

npm install jsonstream streamifier
# yarn用户执行：yarn add jsonstream streamifier

2. 修改NestJS接口代码 #

import { Response } from 'express';
import * as streamifier from 'streamifier';
import * as JSONStream from 'JSONStream';
import { createWriteStream } from 'fs';
import { join } from 'path';

@Post('upload')
@ApiConsumes('multipart/form-data')
@ApiBody({
    description: 'Geojson File',
    type: FileUploadDto,
})
@UseInterceptors(FileInterceptor('file'))
uploadFile(@UploadedFile() file: Express.Multer.File, @Res() res: Response) {
    // 从上传文件的buffer创建可读流
    const inputStream = streamifier.createReadStream(file.buffer);

    // 定位features数组的每个元素，提取geometry和properties
    const transformStream = JSONStream.parse('features.*', (feature) => {
        return {
            geometry: feature.geometry,
            properties: feature.properties
        };
    });

    // 创建可写流，将过滤后的内容保存到指定文件
    const outputFileStream = createWriteStream(join(__dirname, '../filtered-features.json'));

    // 串联流：输入 -> 转换 -> 保存到文件 + 返回给客户端
    const pipeline = inputStream
        .pipe(transformStream)
        .pipe(JSONStream.stringify('[', ',', ']')) // 把单个对象拼成合法JSON数组
        .pipe(outputFileStream)
        .pipe(res);

    // 处理流异常
    pipeline.on('error', (err) => {
        console.error('文件处理出错:', err);
        res.status(500).send('文件处理失败');
    });
}

方案细节 #

• JSONStream.parse('features.*')：精准定位到GeoJSON中features数组的每一项，每次回调处理单个feature对象。
• 字段过滤：在回调里只保留需要的geometry和properties，直接丢弃其他字段。
• JSONStream.stringify：因为提取的是零散的对象，用这个方法把它们包装成合法的JSON数组，避免输出无效JSON。
• 流串联：通过pipe把各个流串起来，实现边读、边处理、边输出，内存占用极低，适合大文件场景。

无第三方库的替代实现 #

如果不想引入第三方库，可以自定义Transform流，但仅适配当前GeoJSON结构，生产环境不推荐：

import { Transform } from 'stream';

class FeatureFilter extends Transform {
    private buffer = '';
    private inFeatures = false;
    private featureBuffer = '';
    private depth = 0;

    constructor() {
        super({ readableObjectMode: true });
    }

    _transform(chunk, encoding, callback) {
        this.buffer += chunk.toString();
        this.parseBuffer();
        callback();
    }

    _flush(callback) {
        this.parseBuffer();
        callback();
    }

    parseBuffer() {
        while (this.buffer.length > 0) {
            if (!this.inFeatures) {
                const featuresMarker = this.buffer.indexOf('"features": [');
                if (featuresMarker === -1) break;
                this.inFeatures = true;
                this.buffer = this.buffer.slice(featuresMarker + 11);
            } else {
                const char = this.buffer[0];
                this.buffer = this.buffer.slice(1);

                if (char === '{') {
                    this.depth++;
                    this.featureBuffer += char;
                } else if (char === '}') {
                    this.depth--;
                    this.featureBuffer += char;
                    if (this.depth === 0) {
                        try {
                            const feature = JSON.parse(this.featureBuffer);
                            this.push({
                                geometry: feature.geometry,
                                properties: feature.properties
                            });
                        } catch (err) {
                            this.emit('error', err);
                        }
                        this.featureBuffer = '';
                    }
                } else if (this.depth > 0) {
                    this.featureBuffer += char;
                } else if (char === ']') {
                    this.inFeatures = false;
                    break;
                }
            }
        }
    }
}

// 使用时替换transformStream为new FeatureFilter()

这种自定义流只针对特定结构的GeoJSON，遇到格式变化容易出错，优先用JSONStream方案。

内容的提问来源于stack exchange，提问作者Angels