如何反转流以从后向前读取JSONL文件？

问题背景 #

我有一个在线JSONL文件，格式示例如下：

{ "id": "something1" }
{ "id": "foo1", "__parentId": "something1" }
{ "id": "something2" }
{ "id": "something3" }
{ "id": "foo2", "__parentId": "something3" }
{ "id": "something4" }
{ "id": "foo3", "__parentId": "something2" }

该文件体积可能极大（甚至达20GB）。目前使用以下代码流式读取：

await axios.get(url, { responseType: "stream" }).then(async (res) => {
  const rl = readline.createInterface({
    input: res.data,
    crlfDelay: Infinity,
  });

  for await (const line of rl) {
    if (line) {
      await handleLine(JSON.parse(line));
    }
  }
});

这段代码可以正常运行，但它是从上到下顺序读取文件的。我需要从下到上解析文件，确保处理父节点时所有子节点已经处理完毕。

曾尝试将文件保存到临时文件后反向读取，这种方法可行，但文件可能无限大，因此希望直接用Node.js流处理，无需存储到本地内存。想知道能否通过TransformStream之类的工具实现这种转换？

解决方案 #

直接反向流式处理HTTP响应流本质上不可行——因为HTTP协议是单向顺序传输的，服务器无法从文件末尾开始发送数据（除非服务器支持范围请求且你能预先获取文件总大小）。不过可以通过以下两种思路解决问题：

思路1：利用HTTP范围请求（需服务器支持） #

如果目标服务器支持HTTP Range 请求头，可以从文件末尾分块拉取数据，再反向解析行内容：

1. 先发送HEAD请求获取文件总字节数
2. 从文件末尾开始，分块请求数据片段
3. 对每个片段进行反向行解析，拼接完整行后从下到上处理

示例代码逻辑：

const axios = require('axios');
const readline = require('readline');
const { Transform } = require('stream');

async function processReverse(url) {
  // 获取文件总大小
  const headRes = await axios.head(url);
  const fileSize = parseInt(headRes.headers['content-length'], 10);
  
  const chunkSize = 1024 * 1024; // 1MB块大小，可根据内存调整
  let start = fileSize - chunkSize;
  let leftover = '';

  while (start >= 0) {
    // 调整起始位置，避免负数
    const actualStart = Math.max(start, 0);
    const res = await axios.get(url, {
      headers: { Range: `bytes=${actualStart}-${start + chunkSize - 1}` },
      responseType: 'stream'
    });

    // 自定义转换流：将二进制数据转字符串，反向提取完整行
    const reverseTransform = new Transform({
      transform(chunk, encoding, callback) {
        let data = chunk.toString('utf8');
        // 拼接上一次剩余的不完整行
        data = leftover + data;
        const lines = data.split('\n');
        // 保存当前块的首行（可能不完整）到下一轮处理
        leftover = lines.shift();
        // 反向遍历当前块的完整行，推入流中
        for (let i = lines.length - 1; i >= 0; i--) {
          if (lines[i]) this.push(lines[i] + '\n');
        }
        callback();
      },
      flush(callback) {
        // 处理最后剩余的完整行
        if (leftover) this.push(leftover + '\n');
        callback();
      }
    });

    // 读取转换后的反向行流
    const rl = readline.createInterface({
      input: res.data.pipe(reverseTransform),
      crlfDelay: Infinity
    });

    for await (const line of rl) {
      if (line) await handleLine(JSON.parse(line));
    }

    start -= chunkSize;
  }
}

注意：该方法依赖服务器支持范围请求，且需要处理行被分割在两个数据块中的情况。

思路2：缓存子节点，延迟处理父节点 #

如果无法使用范围请求，可以在正向流处理时，用两个Map分别缓存节点和子节点关系：遇到子节点先暂存，遇到父节点时先处理所有对应的子节点，再处理父节点；无父节点的根节点最后统一处理。

示例代码：

const axios = require('axios');
const readline = require('readline');

async function processWithCache(url) {
  const nodeCache = new Map(); // 缓存待处理的根/父节点（key: id）
  const childrenMap = new Map(); // 缓存子节点列表（key: parentId）

  await axios.get(url, { responseType: "stream" }).then(async (res) => {
    const rl = readline.createInterface({
      input: res.data,
      crlfDelay: Infinity,
    });

    for await (const line of rl) {
      if (!line) continue;
      const node = JSON.parse(line);
      
      if (node.__parentId) {
        // 子节点：存入对应父节点的子列表
        if (!childrenMap.has(node.__parentId)) {
          childrenMap.set(node.__parentId, []);
        }
        childrenMap.get(node.__parentId).push(node);
      } else {
        // 暂存根节点
        nodeCache.set(node.id, node);
      }

      // 检查当前节点是否有缓存的子节点，有则先处理子节点再处理自身
      if (childrenMap.has(node.id)) {
        for (const child of childrenMap.get(node.id)) {
          await handleLine(child);
        }
        childrenMap.delete(node.id);
        await handleLine(node);
        nodeCache.delete(node.id);
      }
    }
  });

  // 处理剩余无父节点的根节点
  for (const node of nodeCache.values()) {
    await handleLine(node);
  }
}

这种方法无需反向读取，通过缓存机制确保子节点先于父节点处理，内存占用仅取决于未处理的节点数量，远低于缓存整个文件。

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