如何重新调度NetSuite Map Reduce脚本并解决CSV生成难题

核心问题梳理 #

• 首次执行生成空CSV文件：非目标执行日时，脚本仍会创建带表头的空文件
• 重调度追加行受限制：file.load存在2GB大小阈值，大数据量下加载文件追加会触发使用量超限
• Reduce阶段治理超限：当前Reduce逻辑无实际业务价值，且getInputData一次性加载全量数据到内存，直接消耗大量治理单元

针对性修复方案 #

1. 解决首次执行空CSV问题 #

在summarize阶段先校验输出结果数量，无有效数据则跳过文件创建与邮件发送：

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

        if (!recipientEmailsString) {
            log.error('Email Error', 'Recipient emails not provided in script parameter.');
            return;
        }

        let recipientEmailsArray = recipientEmailsString.split(',').map(email => email.trim());
        log.audit("recipientEmailsArray", recipientEmailsArray);

        // 统计输出行数，无数据则直接退出
        let outputCount = 0;
        summary.output.iterator().each(() => {
            outputCount++;
            return true;
        });
        if (outputCount === 0) {
            log.audit('No Data', 'No dataset results to generate report.');
            return;
        }

        // 后续文件创建逻辑...
    } catch (error) {
        log.error('Error in summarize stage', error);
    }
}

2. 优化重调度与大数据量文件生成 #

放弃file.load追加方式，采用分块生成临时文件+最终合并方案，同时实现分页重调度逻辑：

修改getInputData实现分页加载与重调度参数传递

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

        if (today.toDateString() !== firstWorkingDay.toDateString()) {
            log.audit('Not First Working Day', `Today: ${today.toDateString()}, First Working Day: ${firstWorkingDay.toDateString()}`);
            return [];
        }

        let scriptObj = runtime.getCurrentScript();
        let currentPageIndex = scriptObj.getParameter({ name: 'custscript_current_page_index' }) || 0;
        let pageSize = 1000;

        let datasetObj = dataset.load({ id: 'custdataset70' });
        let pagedData = datasetObj.runPaged({ pageSize: pageSize });
        let totalPages = pagedData.pageRanges.length;

        // 当前页超出总页数，返回空表示任务完成
        if (currentPageIndex >= totalPages) {
            log.audit('All Pages Processed', `Total pages: ${totalPages}`);
            return [];
        }

        // 仅加载当前页数据，避免内存溢出
        let currentPage = pagedData.fetch({ index: currentPageIndex });
        let pageResults = currentPage.data.results.map(result => ({
            rowValues: result.values,
            datasetColumns: datasetObj.columns.map(col => ({ id: col.id, alias: col.alias }))
        }));

        log.audit(`Processing Page ${currentPageIndex + 1}/${totalPages}`, `Row count: ${pageResults.length}`);
        return pageResults;
    } catch (error) {
        log.error('Error in getInputData', error);
        return [];
    }
}

在summarize中实现重调度与临时文件合并

function summarize(summary) {
    try {
        let scriptObj = runtime.getCurrentScript();
        let recipientEmailsString = scriptObj.getParameter({ name: 'custscript_recepient_email_address' });
        let currentPageIndex = scriptObj.getParameter({ name: 'custscript_current_page_index' }) || 0;
        let tempFolderId = 5081389; // 临时文件存储文件夹

        if (!recipientEmailsString) {
            log.error('Email Error', 'Recipient emails not provided in script parameter.');
            return;
        }

        let recipientEmailsArray = recipientEmailsString.split(',').map(email => email.trim());

        // 生成当前页临时CSV
        let datasetObj = dataset.load({ id: 'custdataset70' });
        let headers = datasetObj.columns.map(col => col.label || col.id);
        let tempFileName = `Vendor_Spend_Temp_${currentPageIndex}.csv`;
        let tempFileObj = file.create({
            name: tempFileName,
            fileType: file.Type.CSV,
            contents: '',
            isOnline: true,
            folder: tempFolderId
        });

        // 仅第一页添加表头
        if (currentPageIndex === 0) {
            tempFileObj.appendLine({ value: headers });
        }

        summary.output.iterator().each((key, value) => {
            tempFileObj.appendLine({ value: value });
            return true;
        });
        tempFileObj.save();
        log.audit('Temp File Saved', `Page ${currentPageIndex} temp file: ${tempFileName}`);

        // 检查是否需要重调度下一页
        let datasetObjFull = dataset.load({ id: 'custdataset70' });
        let pagedData = datasetObjFull.runPaged({ pageSize: 1000 });
        let totalPages = pagedData.pageRanges.length;

        if (currentPageIndex + 1 < totalPages) {
            runtime.script.instantiate({
                scriptId: scriptObj.id,
                deploymentId: scriptObj.deploymentId,
                parameters: {
                    custscript_current_page_index: currentPageIndex + 1,
                    custscript_recepient_email_address: recipientEmailsString
                }
            });
            log.audit('Rescheduled', `Next page: ${currentPageIndex + 1}/${totalPages}`);
            return;
        }

        // 所有页处理完成，合并临时文件
        let searchObj = search.create({
            type: search.Type.FILE,
            filters: [
                ['folder', 'is', tempFolderId],
                'and',
                ['name', 'contains', 'Vendor_Spend_Temp_']
            ],
            columns: ['internalid']
        });

        let finalFileName = 'Vendor Spend Report_' + new Date(new Date().toLocaleString("en-US", { timeZone: "Australia/Sydney" })).toISOString().replace(/:/g, '_') + '.csv';
        let finalFileObj = file.create({
            name: finalFileName,
            fileType: file.Type.CSV,
            contents: '',
            isOnline: true,
            folder: tempFolderId
        });

        // 按页码顺序合并临时文件
        let tempFileIds = [];
        searchObj.run().each(result => {
            tempFileIds.push(result.id);
            return true;
        });
        tempFileIds.sort((a, b) => {
            let pageA = parseInt(a.split('_')[3]);
            let pageB = parseInt(b.split('_')[3]);
            return pageA - pageB;
        });

        tempFileIds.forEach(fileId => {
            let tempFile = file.load({ id: fileId });
            finalFileObj.appendLine({ value: tempFile.getContents() });
            tempFile.delete(); // 清理临时文件
        });

        // 压缩并发送邮件
        let archiver = compress.createArchiver();
        archiver.add({ file: finalFileObj });
        let zipFile = archiver.archive({
            name: `Vendor Spend Report File_${new Date(new Date().toLocaleString("en-US", { timeZone: "Australia/Sydney" })).toJSON()}.zip`
        });
        zipFile.folder = tempFolderId;
        let zipFileId = zipFile.save();

        if (zipFileId) {
            sendEmail(zipFileId, recipientEmailsArray);
            finalFileObj.delete(); // 保留压缩包，删除原始CSV
        }

        log.audit('Success', 'Email sent with dataset results.');
        log.audit('Remaining governance units', runtime.getCurrentScript().getRemainingUsage());

    } catch (error) {
        log.error('Error in summarize stage', error);
    }
}

3. 解决Reduce阶段治理超限问题 #

直接移除无意义的Reduce阶段，Map阶段输出直接传递到Summary：

// 删除reduce函数，修改返回对象
return {
    getInputData: getInputData,
    map: map,
    summarize: summarize
};

同时修改Map阶段的输出逻辑，避免重复键值：

// Map阶段直接输出行数据，键设为固定值
context.write({ key: 'csv_row', value: csvRow });

额外优化建议 #

• 将lookupField中的单条查找改为批量查找，减少治理单元消耗
• 脚本参数中添加临时文件夹ID配置，避免硬编码
• 增加临时文件清理的容错机制，防止残留文件占用空间

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