NetSuite Map Reduce脚本Reduce阶段使用量超限问题求助

问题根源 #

你的脚本Reduce阶段无实际聚合逻辑，仅传递Map输出的单条数据，导致大量独立Reduce任务被创建，每个任务都消耗治理单元，数据量大时直接触发超限。同时，getInputData全量加载数据集到内存、Map阶段频繁单条lookup操作也加剧了治理消耗。

优化方案 #

1. 移除冗余的Reduce阶段 #

当前Reduce阶段仅做数据传递，无任何聚合/合并操作，完全可以跳过该阶段，让Map输出直接进入Summarize阶段处理，减少治理单元浪费。

2. 优化getInputData，避免内存过载 #

原代码将所有数据集结果存入数组后返回，数据量大时会占用大量内存且增加框架处理负担。改为直接返回数据集对象，让NetSuite自动分页处理输入数据。

3. 缓存Lookup结果，减少API调用 #

原Map阶段每条数据都执行多次单条lookup，每个lookup都会消耗治理单元。新增缓存机制，同一个记录ID只查询一次，大幅减少API调用次数。

4. 批量写入CSV，提升效率 #

原代码逐行appendLine写入文件，数据量大时IO操作频繁。改为先将所有行存入数组，最后一次性写入文件内容，提升文件生成效率。

修改后的完整代码 #

/**
* @NApiVersion 2.1
* @NScriptType MapReduceScript
* Script Description:
* 该Map/Reduce脚本提取数据集结果，生成CSV文件并发送给指定收件人，优化后支持处理大数据量。
*/
define(['N/email', 'N/file', 'N/runtime', 'N/dataset', 'N/log', 'N/search', 'N/compress'],
function (email, file, runtime, dataset, log, search, compress) {

    // 预加载需要的名称映射，减少重复lookup次数
    let nameMappings = {
        customer: new Map(),
        vertical: new Map(),
        subdepartment: new Map(),
        subsidiary: new Map(),
        currency: new Map()
    };

    function getInputData() {
        try {
            let today = new Date();
            let firstWorkingDay = getFirstWorkingDay(today.getFullYear(), today.getMonth() + 1);

            // 非当月第一个工作日则退出
            if (today.toDateString() !== firstWorkingDay.toDateString()) {
                log.audit('非第一个工作日', `今日: ${today.toDateString()}, 当月第一个工作日: ${firstWorkingDay.toDateString()}`);
                return [];
            }

            // 直接返回数据集对象，让框架自动分页处理
            let datasetObj = dataset.load({ id: 'custdataset70' });
            log.audit('数据集加载完成', `列数: ${datasetObj.columns.length}`);
            return datasetObj;
        } catch (error) {
            log.error('getInputData错误', error);
            return [];
        }
    }

    function map(context) {
        try {
            let result = context.value;
            let rowValues = result.values;
            let datasetColumns = context.columns.map(col => ({ id: col.id, alias: col.alias }));

            let poNumberIndex = -1, poStatusIndex = -1, billStatusIndex = -1, currencyIndex = -1;
            let customerIndex = -1, verticalIndex = -1, subproductIndex = -1, subsidiaryIndex = -1;

            // 获取列索引
            datasetColumns.forEach((col, index) => {
                if (col.alias === 'tranid') poNumberIndex = index;
                if (col.alias === 'status') poStatusIndex = index;
                if (col.alias === 'status_1') billStatusIndex = index;
                if (col.alias === 'currency') currencyIndex = index;
                if (col.alias === 'entity') customerIndex = index;
                if (col.alias === 'cseg_vertical') verticalIndex = index;
                if (col.alias === 'custbody_subdepartment') subproductIndex = index;
                if (col.alias === 'subsidiary') subsidiaryIndex = index;
            });

            if (poNumberIndex === -1 || poStatusIndex === -1 || billStatusIndex === -1 || currencyIndex === -1 || customerIndex === -1 || verticalIndex === -1 || subproductIndex === -1 || subsidiaryIndex === -1) {
                log.error('错误', '数据集中未找到必要列');
                return;
            }

            // 替换账单状态
            if (billStatusIndex !== -1) {
                switch(rowValues[billStatusIndex]) {
                    case 'A': rowValues[billStatusIndex] = 'Bill: Open'; break;
                    case 'B': rowValues[billStatusIndex] = 'Bill: Paid In Full'; break;
                    case 'C': rowValues[billStatusIndex] = 'Item Fulfillment : Shipped'; break;
                    case 'Y': rowValues[billStatusIndex] = 'Item Receipt : Undefined'; break;
                }
            }

            // 替换采购订单状态
            if (poStatusIndex !== -1) {
                switch(rowValues[poStatusIndex]) {
                    case 'A': rowValues[poStatusIndex] = 'Purchase Order: Pending Supervisor Approval'; break;
                    case 'B': rowValues[poStatusIndex] = 'Purchase Order: Pending Receipt'; break;
                    case 'C': rowValues[poStatusIndex] = 'Purchase Order : Rejected by Supervisor'; break;
                    case 'D': rowValues[poStatusIndex] = 'Purchase Order : Partially Received'; break;
                    case 'E': rowValues[poStatusIndex] = 'Purchase Order : Pending Billing/Partially Received'; break;
                    case 'F': rowValues[poStatusIndex] = 'Purchase Order : Pending Bill'; break;
                    case 'G': rowValues[poStatusIndex] = 'Purchase Order : Fully Billed'; break;
                    case 'H': rowValues[poStatusIndex] = 'Purchase Order : Closed'; break;
                }
            }

            // 处理客户信息，使用缓存避免重复查询
            if(rowValues[customerIndex]) {
                let customerId = rowValues[customerIndex];
                if (!nameMappings.customer.has(customerId)) {
                    let customerLookup = search.lookupFields({
                        type: search.Type.CUSTOMER,
                        id: customerId,
                        columns: ['entityid', 'altname']
                    });
                    let concatnatecustomer = `${customerLookup.entityid} ${customerLookup.altname}`;
                    nameMappings.customer.set(customerId, concatnatecustomer);
                }
                rowValues[customerIndex] = nameMappings.customer.get(customerId);
            }

            // 处理其他lookup字段，使用缓存
            rowValues[verticalIndex] = getCachedName('vertical', 'customrecord_cseg_vertical', rowValues[verticalIndex]);
            rowValues[subproductIndex] = getCachedName('subdepartment', 'customrecord_subdepartment', rowValues[subproductIndex]);
            rowValues[subsidiaryIndex] = getCachedName('subsidiary', 'subsidiary', rowValues[subsidiaryIndex]);
            rowValues[currencyIndex] = getCachedName('currency', 'currency', rowValues[currencyIndex]);

            // 格式化CSV行
            let csvRow = rowValues.map(value => {
                if (typeof value === 'string' && value.includes(',')) {
                    return `"${value.replace(/"/g, '""')}"`;
                }
                return value ?? '';
            }).join(',');

            context.write({ key: 'csv_row', value: csvRow });
        } catch (error) {
            log.error('Map阶段错误', JSON.stringify(error));
        }
    }

    function summarize(summary) {
        try {
            let scriptObj = runtime.getCurrentScript();
            let recipientEmailsString = scriptObj.getParameter({ name: 'custscript_recepient_email_address' });

            if (!recipientEmailsString) {
                log.error('邮件错误', '未在脚本参数中提供收件人邮箱');
                return;
            }

            let recipientEmailsArray = recipientEmailsString.split(',').map(email => email.trim());
            log.audit('收件人列表', recipientEmailsArray);

            // 获取CSV表头
            let datasetObj = dataset.load({ id: 'custdataset70' });
            let headers = datasetObj.columns.map(col => col.label || col.id);
            let csvRows = [headers.join(',')];

            // 收集所有CSV行
            summary.output.iterator().each((key, value) => {
                csvRows.push(value);
                return true;
            });

            // 生成CSV文件
            let today = new Date(new Date().toLocaleString("en-US", { timeZone: "Australia/Sydney" }));
            let isoString = today.toISOString().replace(/:/g, '_');
            let fileName = `Vendor Spend Report_${isoString}.csv`; 
            let fileObj = file.create({
                name: fileName,
                fileType: file.Type.CSV,
                contents: csvRows.join('\n'),
                isOnline: true,
                folder: 5081389
            });

            // 压缩文件
            let archiver = compress.createArchiver();
            archiver.add({ file: fileObj });
            let zipFileName = `Vendor Spend Report File_${new Date(today).toJSON()}.zip`;
            let zipFile = archiver.archive({ name: zipFileName });
            zipFile.folder = 5081389; 
            let fileId = zipFile.save();

            if (fileId) {
                sendEmail(fileId, recipientEmailsArray);
            }

            log.audit('成功', '已发送包含数据集结果的邮件');
            log.audit('剩余治理单元', runtime.getCurrentScript().getRemainingUsage());
        } catch (error) {
            log.error('Summarize阶段错误', error);
        }
    }

    function sendEmail(fileId, recipientEmailsArray) {
        try {
            if (!fileId) {
                log.audit('邮件跳过', '未生成文件');
                return;
            }

            let lastDayOfPreviousMonth = getLastDayOfPreviousMonth("Australia/Sydney");
            if (!lastDayOfPreviousMonth) return;

            let startDate = '1 January 2022';
            let endDate = lastDayOfPreviousMonth.toLocaleDateString('en-AU', { day: 'numeric', month: 'long', year: 'numeric' });

            let emailBody = `<p>*** 这是自动发送的邮件，请不要回复。 ****</p>` +
                `<p>亲爱的用户：</p>` +
                `<p>请查收附件中2022年1月1日至${endDate}的供应商支出报告。</p>` +
                `<p>谢谢<br>` + 
                `NetSuite 支持团队</p>` +
                `<p>请勿直接回复此邮件，我们无法处理此类回复。若您并非该邮件的合适收件人，请通过NetSuite支持联系我们：<br>` +
                `netsuitesupport@team.telstra.com</p>`;

            email.send({
                author: 115293, 
                recipients: recipientEmailsArray, 
                subject: '月度供应商支出报告',
                body: emailBody,
                attachments: [file.load({ id: fileId })]
            });
        } catch (error) {
            log.error('发送邮件错误', JSON.stringify(error));
        }
    }

    // 缓存lookup结果，减少重复API调用
    function getCachedName(mapKey, recordType, internalId) {
        if (!internalId) return '';
        let map = nameMappings[mapKey];
        if (map.has(internalId)) {
            return map.get(internalId);
        }
        try {
            let lookup = search.lookupFields({
                type: recordType,
                id: internalId,
                columns: ['name']
            });
            let name = lookup.name ?? 'Null';
            map.set(internalId, name);
            return name;
        } catch (error) {
            log.error(`查询${recordType} ID ${internalId}错误`, JSON.stringify(error));
            return 'Null';
        }
    }

    function getFirstWorkingDay(year, month) {
        let date = new Date(year, month - 1, 1);
        while (date.getDay() === 0 || date.getDay() === 6) { // 0=周日，6=周六
            date.setDate(date.getDate() + 1);
        }
        return date;
    }

    function getLastDayOfPreviousMonth(timezone) {
        try {
            let todayDate = new Date(new Date().toLocaleString("en-US", { timeZone: timezone }));
            return new Date(todayDate.getFullYear(), todayDate.getMonth(), 0);
        } catch (error) {
            log.error('获取上月最后一天错误', JSON.stringify(error));
            return null;
        }
    }

    return {
        getInputData: getInputData,
        map: map,
        summarize: summarize // 移除reduce阶段
    };
});

关键优化点说明 #

• 移除Reduce阶段：减少不必要的任务调度和治理消耗，Map输出直接进入Summarize。
• 缓存Lookup结果：同一个记录ID只查询一次，大幅减少API调用次数。
• 批量写入CSV：先收集所有行再一次性写入，提升文件生成效率。
• 直接返回数据集：让NetSuite框架自动分页处理输入，避免内存过载。

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