如何从NodeJS将BigQuery返回的JSON字符串上传至GCS？

问题场景 #

从BigQuery拉取结果集并遍历，需将每行中包含JSON字符串的列上传至GCS存储桶作为独立文件。JSON示例：

{
"name":"sharath",
"country":"India"
}

尝试使用file.save()和PassThrough流两种方式均无效果，无报错但GCS未生成任何文件。

核心问题分析 #

1. 文件名非法/重复：代码中使用*.json作为文件名，GCS不允许文件名包含通配符*；循环中若文件名重复，会导致文件无法正常创建或被覆盖。
2. 异步操作未等待：myFunction为async函数，但循环中调用file.save()仅用.then()未加await，函数执行完毕时上传操作尚未完成就退出，导致上传未触发完成。
3. 路径格式错误：GCS文件路径无需以/开头，开头的斜杠会导致路径解析异常。
4. 流操作未处理异步完成：使用流时未等待finish事件，循环快速创建多个流可能导致资源未正确处理。

修正方案 #

方案1：使用file.save()（推荐） #

通过Promise.all等待所有上传任务完成，同时生成唯一文件名：

const xxx = { 
  myFunction: async () => {
    try{
      // ...从BigQuery获取数据的逻辑...
      
      // 批量处理所有行，等待全部上传完成
      await Promise.all(rows.map(async (row, index) => {
        const contents = row.JSON_Content;
        // 生成唯一文件名（示例用行索引，可替换为业务唯一ID）
        const fileName = `data_${index}.json`;
        // 拼接合法的GCS文件路径（无开头斜杠）
        const filePath = `${process.env.FILE_TMP_PATH}/${fileName}`;
        const file = storage.bucket(gcsBucket).file(filePath);
        
        await file.save(contents);
        console.log(`文件 ${filePath} 上传完成`);
      }));
      
    }catch(err){
      throw new Error(`操作失败: ${err.message}`);
    }
  }
}
module.exports = xxx;

方案2：使用流上传 #

将流操作包装为Promise，确保等待上传完成：

const stream = require('stream');

const xxx = { 
  myFunction: async () => {
    try{
      // ...从BigQuery获取数据的逻辑...
      
      await Promise.all(rows.map(async (row, index) => {
        const contents = row.JSON_Content;
        const fileName = `data_${index}.json`;
        const filePath = `${process.env.FILE_TMP_PATH}/${fileName}`;
        const file = storage.bucket(gcsBucket).file(filePath);
        
        // 将流操作包装为Promise，等待finish/error事件
        return new Promise((resolve, reject) => {
          const passthroughStream = new stream.PassThrough();
          passthroughStream.write(contents);
          passthroughStream.end();
          
          passthroughStream.pipe(file.createWriteStream())
            .on('error', (err) => reject(new Error(`文件上传失败: ${err.message}`)))
            .on('finish', () => {
              console.log(`文件 ${filePath} 上传完成`);
              resolve();
            });
        });
      }));
      
    }catch(err){
      throw new Error(`操作失败: ${err.message}`);
    }
  }
}
module.exports = xxx;

额外注意事项 #

• 确保process.env.FILE_TMP_PATH是GCS桶内的有效路径（如temp/json_files），不要以/开头。
• 若row.JSON_Content是JSON对象而非字符串，需先通过JSON.stringify(contents)转换为字符串再上传。
• 验证服务账号权限：确保运行代码的账号拥有GCS存储桶的storage.objects.create权限。

内容的提问来源于stack exchange，提问作者shary.sharath