如何将Fetch返回的大JSON响应转为流并接入big-json处理？

问题背景 #

我的应用需要获取大于600MB的大JSON数据，可根据配置从本地文件或API获取。由于JSON字符串超出允许大小，无法使用JSON.parse解析，因此计划用big-json库处理。本地文件部分已实现返回fs.createReadStream，但API部分不知道如何将Fetch的响应体转为可pipe到big-json.createParseStream()的流，需要完善getDataFromApi方法。

现有代码示例：

import fetch from 'node-fetch';
const json = require('big-json');
// 注意：需要导入fs模块
import fs from 'fs';

const myDataStream = await getMyDataStream(mySettings);
const parseStream = json.createParseStream();

parseStream.on('data', function(pojo) {
    // => 后续逻辑
});

myDataStream.pipe(parseStream);

const getMyDataStream = async ({queryId, myFile, myApiEndpoint}) => {
  if (myFile)
    return fs.createReadStream('myFile')
  else
    // 注意：这里拼写要和函数名一致，原代码里是getDataFromApi
    return await getDataFromApi({queryId, myApiEndpoint});
}

const getDataFromApi = async ({queryId, myApiEndpoint}) => {
  const response = await fetch(
    `${myApiEndpoint}/queries/${queryId}`,
    {
      method: 'GET',
      compress: true,
    }
  )
  return response.body; // 返回响应体流
}

解决方案 #

实际上node-fetch返回的response.body本身就是Node.js标准的Readable流，可以直接pipe到big-json.createParseStream()，但需要补充几个关键细节：

1. 校验响应状态：确保API返回成功状态码，避免将错误响应（如404、500）的内容当作JSON解析
2. 明确请求格式：添加Accept头确保API返回JSON格式数据
3. 修复拼写错误：原代码中getMyDataStream里调用的getDataFromAPI拼写错误，应改为getDataFromApi
4. 导入缺失模块：补充fs模块的导入

完善后的完整代码：

import fetch from 'node-fetch';
const json = require('big-json');
import fs from 'fs';

const myDataStream = await getMyDataStream(mySettings);
const parseStream = json.createParseStream();

parseStream.on('data', function(pojo) {
    // 处理解析后的JSON片段或完整对象
    console.log('解析得到数据片段:', pojo);
});

parseStream.on('error', function(err) {
    // 处理解析错误
    console.error('JSON解析失败:', err);
});

myDataStream.pipe(parseStream);

const getMyDataStream = async ({queryId, myFile, myApiEndpoint}) => {
  if (myFile) {
    return fs.createReadStream(myFile); // 使用传入的myFile参数而非固定字符串
  } else {
    return await getDataFromApi({queryId, myApiEndpoint});
  }
}

const getDataFromApi = async ({queryId, myApiEndpoint}) => {
  const response = await fetch(
    `${myApiEndpoint}/queries/${queryId}`,
    {
      method: 'GET',
      compress: true,
      headers: {
        'Accept': 'application/json' // 明确要求返回JSON格式
      }
    }
  );

  // 校验响应状态
  if (!response.ok) {
    throw new Error(`API请求失败: ${response.status} ${response.statusText}`);
  }

  // 直接返回响应体流，big-json可直接处理Node.js Readable流
  return response.body;
}

关键说明 #

• big-json的createParseStream()支持接收任意Node.js Readable流，node-fetch的response.body完全符合要求
• 添加流的error事件监听，可捕获请求或解析过程中的异常
• 本地文件流使用传入的myFile参数，增强代码灵活性

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