使用createWriteStream流式传输JSON至BigQuery表：无数据写入问题排查

嘿，我碰到过好几个类似的情况——程序跑起来没报错，但BigQuery表里就是没数据，大概率是流的处理逻辑、数据格式或者权限的问题，咱们一步步来排查解决：

1. 确保流的结束逻辑没遗漏 #

BigQuery的流式写入流（比如createWriteStream生成的流）需要明确触发结束动作，否则最后一批数据可能会卡在缓冲区里没提交。如果你的代码只是简单把JSON流pipe过去，没处理流的结束事件，就容易出现这种情况。

给你一个标准的正确示例：

const { BigQuery } = require('@google-cloud/bigquery');
const bigquery = new BigQuery();
const fs = require('fs');
const { Transform } = require('stream');
const JSONStream = require('jsonstream'); // 处理大JSON数组必备

// 你的数据转换流
const transformStream = new Transform({
  objectMode: true,
  transform(chunk, encoding, callback) {
    // 这里做你的轻微转换，比如类型修正、字段重命名
    const transformed = {
      ...chunk,
      // 比如把字符串类型的数字转成Number，匹配BigQuery的INTEGER类型
      age: chunk.age ? Number(chunk.age) : null
    };
    callback(null, transformed);
  }
});

// 创建BigQuery写入流，关键配置不能少
const writeStream = bigquery
  .dataset('你的数据集名称')
  .table('你的目标表名称')
  .createWriteStream({
    schema: { fields: [/* 你的表schema */] },
    // 调整批处理阈值，避免数据卡在缓冲区
    maxRows: 1000,
    maxBytes: 1024 * 1024,
    // 关闭静默忽略，有问题直接报错，方便排查
    ignoreUnknownValues: false,
    skipInvalidRows: false
  });

// 一定要监听错误！不然流的错误会被悄悄吞掉
writeStream.on('error', (err) => {
  console.error('BigQuery写入出错:', err);
});

// 监听完成事件，确认所有数据都提交了
writeStream.on('finish', () => {
  console.log('所有数据已成功写入BigQuery');
});

// 完整的流管道：读取JSON文件 -> 解析JSON数组 -> 转换数据 -> 写入BigQuery
fs.createReadStream('testfile.json')
  .pipe(JSONStream.parse('*')) // 解析JSON数组里的每个对象
  .pipe(transformStream)
  .pipe(writeStream);

2. 检查数据与schema的匹配度 #

BigQuery对数据类型要求非常严格，哪怕你创建表时schema正确，转换后的数据如果类型不匹配，会被静默丢弃（默认配置下），不会抛出错误。比如：

• schema定义是INTEGER，你传了字符串"123"而不是数字123
• DATE字段用了"2023/10/01"而不是BigQuery要求的"2023-10-01"格式
• 嵌套字段的结构和schema不一致

解决办法就是开启上面代码里的ignoreUnknownValues: false和skipInvalidRows: false，这样不符合要求的数据会直接抛出错误，帮你快速定位问题。

3. 确认权限是否足够 #

虽然你能成功创建表，但流式写入需要单独的bigquery.tables.insertAll权限。如果你的服务账号只有表创建/修改权限，没有插入权限，就会出现程序跑通但数据写不进去的情况。

可以给服务账号添加BigQuery Data Editor或者BigQuery Streaming Insert角色，确保权限到位。

4. 留意BigQuery的流式缓冲区延迟 #

流式写入的数据会先进入BigQuery的缓冲区，通常需要几秒到几分钟才会同步到表的查询结果里。如果刚跑完程序就去查，可能还没显示出来。可以去BigQuery控制台的表详情页，查看「流式缓冲区」的状态，看看有没有未提交的数据。

5. 确认JSON文件的读取逻辑正确 #

如果你的JSON文件是一个大数组，直接用fs.createReadStream读取会得到整个字符串，无法被BigQuery写入流处理。必须用JSONStream这类库来逐对象解析，就像上面示例里的JSONStream.parse('*')那样。

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