如何降低处理大文件流分片上传的Node.js Lambda内存占用？

问题背景 #

现有流程为S3存储桶上传文件时触发Lambda，将CSV文件转换为JSON。因文件最大可达数GB，全程采用流处理。当前功能已实现，但测试120MB的CSV文件时，Lambda内存消耗超950MB，内存占用过高。目前采用逐行读取而非分块读取，需优化方案及剩余优化空间。推测内存过高源于PassThrough流的readableHighWaterMark设为1GB，但移除该设置后Lambda会崩溃。

核心问题分析 #

• 流链路冗余且未利用背压：当前用readline.createInterface逐行读取后手动写入transformStream，未依托Node.js流的原生背压机制，导致数据在内存中堆积。
• PassThrough流配置极端：1GB的readableHighWaterMark会让流在内存中缓存大量数据，直接推高内存占用；默认值下崩溃，是因为上传流速度跟不上读取/转换速度，未正确处理背压。
• S3 Upload队列配置不合理：queueSize:4可能导致并发上传的块在内存中堆积，尤其是转换速度快于上传速度时。

优化方案 #

1. 简化流链路，利用原生管道与背压 #

直接将S3读取流通过管道接入transformStream，再接入上传流，去掉readline.createInterface的手动写入逻辑，让Node.js自动处理背压，避免内存堆积。

2. 合理配置PassThrough流（或直接移除） #

若必须保留PassThrough，将readableHighWaterMark调整为合理值（如64MB或128MB），确保全链路背压生效；若无需额外处理，可直接移除PassThrough，将transformStream直接作为上传流的Body。

3. 优化S3 Upload配置 #

• 降低queueSize（如设为2），减少并发上传的块内存占用；
• 确保Upload的流处理正确响应背压，避免数据积压。

4. 优化transformStream的_transform实现 #

确保_transform方法处理完数据后及时调用callback，不要同步阻塞或缓存过多数据，保证流的顺畅流动。

修改后的核心代码示例 #

convertFile函数优化版 #

export const convertFile = async (params: ConversionParams): Promise<boolean> => {
    try {    
        // 获取源文件流
        const readStream = await getStreamedFile({ keyName: params.objectKey });
        // CSV转JSON转换流
        const transformStream = new JsonTransformLineStream({});
    
        // S3上传配置，直接使用transformStream作为Body
        const upload = getUpload({
            keyName: params.objectKey,
            streamBody: transformStream
        });

        return new Promise<boolean>((resolve, reject) => {
            // 直接管道连接，自动处理背压
            readStream.pipe(transformStream);
        
            // 监听流错误
            readStream.on("error", (error) => {
                console.error("读取流错误:", error);
                reject(false);
            });
        
            transformStream.on("error", (error) => {
                console.error("转换流错误:", error);
                reject(false);
            });
        
            // 处理上传完成/失败
            upload.done()
                .then(() => {
                    console.log("上传完成");
                    resolve(true);
                })
                .catch((uploadError) => {
                    console.error("上传错误:", uploadError);
                    reject(false);
                });
        });
    } catch (e) {
        console.error("文件转换错误:", e);
        throw e;
    }
};

getUpload函数调整（移除PassThrough依赖） #

export const getUpload = (params: UploadParams): Upload => {
    try {
        let extension = "json";
        const file = params.fileName ?? params.keyName;
        const [fileName, ext] = file.split(".");
        if (params.fileName) {
            extension = ext;
        }
        const uploadKeyName = `${fileName}.${extension}`;
        const contentType = getMimeType(uploadKeyName);
        const s3params = {
            Bucket: process.env.S3_BUCKET,
            Key: uploadKeyName,
            Body: params.streamBody,
            ContentType: contentType
        };
        const upload = new Upload({
            client: new S3Client({}),
            queueSize: 2, // 降低队列大小减少内存占用
            params: s3params
        });
        const defaultProgress = (progress: Progress) => {
            console.log(`已上传分片: ${progress.part}`);
            console.log(`已加载: ${progress.loaded}`);
            console.log(`总大小: ${progress.total}`);
        };
        upload.on("httpUploadProgress", params.onUploadProgress || defaultProgress);
        return upload;
    } catch (e) {
        console.error(e);
        throw e;
    }
};

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