Cloud Function运行缓慢超出内存上限，BigQuery导出GCS如何优化

问题根因 #

• 调用getQueryResults()会一次性拉取当前id的所有查询结果到内存，单id数据量过大会直接占满内存，这也是你调高内存配置仍然OOM的核心原因：如果单个id的数据集超过你配置的内存阈值，无论怎么调高内存都会触发溢出。
• CSV转换也是全量处理，进一步放大内存占用。
• 113次查询+写入全串行执行，总耗时被拉长。

方案1：改造为全链路流式处理（最低内存占用） #

全程不缓存全量数据，边查询边转格式边写入GCS，内存占用稳定在几十MB级别，不管单id数据量多大都不会OOM，同时增加并发控制提升处理速度。
优化后代码：

const { BigQuery } = require("@google-cloud/bigquery");
const { Storage } = require("@google-cloud/storage");
const { Transform } = require('stream');
const { pipeline } = require('stream/promises');
const { Parser } = require("json2csv");
const bucketName = "xxxx";

const bigquery = new BigQuery();
const storage = new Storage();
const fields = [
  "id",
  "product_name",
  "product_desc",
  "etc"
];
// 并发数可根据实际配额调整，避免触发BigQuery并发限制
const CONCURRENCY_LIMIT = 5;

exports.importBQToGCS = async (req, res) => {
  try {
    const liveMerchantCount = 113;
    const tasks = [];
    for (let i = 1; i < liveMerchantCount; i++) {
      tasks.push(() => processSingleMerchant(i));
    }
    await runWithConcurrency(tasks, CONCURRENCY_LIMIT);
    res.status(200).send("处理完成");
  } catch (err) {
    console.error("全局错误", err);
    res.status(500).send(err.message);
  }
};

async function processSingleMerchant(merchantId) {
  console.log(`开始处理商户${merchantId}`);
  // 参数化查询避免SQL注入
  const query = `SELECT * FROM \`table_name\` WHERE id_number = @merchantId`;
  const options = {
    query,
    location: "EU",
    params: { merchantId }
  };
  // 直接获取查询结果流，不缓存全量数据
  const [job] = await bigquery.createQueryJob(options);
  const bqStream = job.getQueryResultsStream();
  
  // 流式转换CSV格式，不需要等全量结果
  const json2csvParser = new Parser({ fields, header: true });
  const csvTransform = new Transform({
    objectMode: true,
    transform(row, _, callback) {
      callback(null, json2csvParser.parse(row) + '\n');
    }
  });

  // GCS写入流
  const gcsFile = storage.bucket(bucketName).file(`test_${merchantId}.csv`);
  const writeStream = gcsFile.createWriteStream({
    resumable: false,
    validation: false,
    metadata: { "Cache-Control": "public, max-age=31536000" },
  });

  // pipeline自动管理流生命周期，避免内存泄漏
  await pipeline(bqStream, csvTransform, writeStream);
  console.log(`商户${merchantId}处理完成`);
}

// 并发控制工具函数
async function runWithConcurrency(tasks, limit) {
  const executing = [];
  for (const task of tasks) {
    const p = Promise.resolve().then(() => task());
    executing.push(p);
    if (executing.length >= limit) {
      const finished = await Promise.race(executing);
      executing.splice(executing.indexOf(finished), 1);
    }
  }
  return Promise.all(executing);
}

方案2：使用BigQuery原生EXPORT（优先推荐，最快最省资源） #

不需要自定义处理数据流转，直接提交BigQuery导出任务让云原生能力处理，导出速度比自定义代码快10倍以上，Cloud Function仅需等待任务完成，内存占用可以忽略不计。
核心逻辑代码片段：

async function exportWithBQNative(merchantId) {
  const exportQuery = `
    EXPORT DATA
    OPTIONS(
      uri = 'gs://${bucketName}/test_${merchantId}.csv',
      format = 'CSV',
      overwrite = true,
      header = true,
      field_delimiter = ','
    ) AS
    SELECT * FROM \`table_name\` WHERE id_number = @merchantId
  `;
  const options = {
    query: exportQuery,
    location: "EU",
    params: { merchantId }
  };
  // 等待原生导出任务完成即可
  const [job] = await bigquery.createQueryJob(options);
  await job.getQueryResults();
  console.log(`商户${merchantId}原生导出完成`);
}

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