javascript - @aws-sdk/lib-storage 使用 JSONStream.stringify() 将 JSON 从 MongoDB 流式传输到 S3

Question

我正在尝试使用新版本的@aws-sdk/lib-storage 将 JSON 从 MongoDB 流式传输到 S3:

"@aws-sdk/client-s3": "^3.17.0"
"@aws-sdk/lib-storage": "^3.34.0"
"JSONStream": "^1.3.5",

尝试#1:看来我没有正确使用 JSONStream.stringify() :

import { MongoClient } from 'mongodb';
import { S3Client } from '@aws-sdk/client-s3';
import { Upload } from '@aws-sdk/lib-storage';
const s3Client = new S3Client({ region: env.AWS_REGION });

export const uploadMongoStreamToS3 = async (connectionString, collectionName) => {
  let client;

  try {
    client = await MongoClient.connect(connectionString);
    const db = client.db();
    const readStream = db.collection(collectionName).find('{}').limit(5).stream();
    readStream.pipe(JSONStream.stringify());
 
    const upload = new Upload({
      client: s3Client,
      params: {
        Bucket: 'test-bucket',
        Key: 'extracted-data/benda_mongo.json',
        Body: readStream,
      },
    });
    
    await upload.done(); 
  }
  catch (err) {
    log.error(err);
    throw err.name;
  }
  finally {
    if (client) {
      client.close();
    }
  }

};

错误 #1:

TypeError [ERR_INVALID_ARG_TYPE]: The first argument must be one oftype string, Buffer, ArrayBuffer, Array, or Array-like Object.Received type objectat Function.from (buffer.js:305:9)at getDataReadable (/.../node_modules/@aws-sdk/lib-storage/src/chunks/getDataReadable.ts:6:18)at processTicksAndRejections (internal/process/task_queues.js:94:5)at Object.getChunkStream (/.../node_modules/@aws-sdk/lib-storage/src/chunks/getChunkStream.ts:17:20)at Upload.__doConcurrentUpload (/.../node_modules/@aws-sdk/lib-storage/src/Upload.ts:121:22)at async Promise.all (index 0)at Upload.__doMultipartUpload (/.../node_modules/@aws-sdk/lib-storage/src/Upload.ts:196:5)at Upload.done (/.../node_modules/@aws-sdk/lib-storage/src/Upload.ts:88:12)

尝试 #2，使用变量 jsonStream :

  const readStream = db.collection(collectionName).find('{}').limit(5).stream();
    const jsonStream = readStream.pipe(JSONStream.stringify());
 
    const upload = new Upload({
      client: s3Client,
      params: {
        Bucket: 'test-bucket',
        Key: 'extracted-data/benda_mongo.json',
        Body: jsonStream,
      },
    });

错误 #2:

ReferenceError: ReadableStream is not definedat Object.getChunk (/.../node_modules/@aws-sdk/lib-storage/src/chunker.ts:22:30)at Upload.__doMultipartUpload (/.../node_modules/@aws-sdk/lib-storage/src/Upload.ts:187:24)at Upload.done (/.../node_modules/@aws-sdk/lib-storage/src/Upload.ts:88:37)

尝试 #3:使用 stream.PassThrough :

    client = await MongoClient.connect(connectionString);
    const db = client.db();
    const readStream = db.collection(collectionName).find('{}').limit(5).stream();
    readStream.pipe(JSONStream.stringify()).pipe(uploadStreamFile('benda_mongo.json'));

...

const stream = require('stream');
export const uploadStreamFile = async(fileName) => {
  try{

    const pass = new stream.PassThrough();
    const upload = new Upload({
      client: s3Client,
      params: {
        Bucket: 'test-bucket',
        Key: 'extracted-data/benda_mongo.json',
        Body: pass,
      },
    });
    const res = await upload.done();
    
    log.info('finished uploading file', fileName);
    return res;
  }
  catch(err){
    return;
  }
};

错误 #3:

'dest.on is not a function at Stream.pipe (internal/streams/legacy.js:30:8'

尝试#4:mongodb.stream({transform: doc => JSON.stringify...}) 而不是 JSONStream:

import { S3Client } from '@aws-sdk/client-s3';
import { Upload } from '@aws-sdk/lib-storage';
import { env } from '../../../env';
const s3Client = new S3Client({ region: env.AWS_REGION });

export const uploadMongoStreamToS3 = async (connectionString, collectionName) => {
  let client;

  try {
    client = await MongoClient.connect(connectionString);
    const db = client.db();
    const readStream = db.collection(collectionName)
      .find('{}')
      .limit(5)
      .stream({ transform: doc => JSON.stringify(doc) + '\n' });
  
    const upload = new Upload({
      client: s3Client,
      params: {
        Bucket: 'test-bucket',
        Key: 'extracted-data/benda_mongo.json',
        Body: readStream,
      },
    });
  
    await upload.done(); 
  }
  catch (err) {
    log.error('waaaaa', err);
    throw err.name;
  }
  finally {
    if (client) {
      client.close();
    }
  }
};

错误:#4:

TypeError [ERR_INVALID_ARG_TYPE]: The first argument must be one oftype string, Buffer, ArrayBuffer, Array, or Array-like Object.Received type objectat Function.from (buffer.js:305:9)at getDataReadable (/.../node_modules/@aws-sdk/lib-storage/src/chunks/getDataReadable.ts:6:18)at processTicksAndRejections (internal/process/task_queues.js:94:5)at Object.getChunkStream (/.../node_modules/@aws-sdk/lib-storage/src/chunks/getChunkStream.ts:17:20)at Upload.__doConcurrentUpload (/.../node_modules/@aws-sdk/lib-storage/src/Upload.ts:121:22)at async Promise.all (index 0)at Upload.__doMultipartUpload (/.../node_modules/@aws-sdk/lib-storage/src/Upload.ts:196:5)at Upload.done (/.../node_modules/@aws-sdk/lib-storage/src/Upload.ts:88:12)

尝试 #5:使用 stream.PassThrough()并返回 pass至 pipe :

export const uploadMongoStreamToS3 = async (connectionString, collectionName) => {
  let client;

  try {
    client = await MongoClient.connect(connectionString);
    const db = client.db();
    const readStream = db.collection(collectionName).find('{}').limit(5).stream({ transform: doc => JSON.stringify(doc) + '\n' });
    readStream.pipe(uploadStreamFile());
  }
  catch (err) {
    log.error('waaaaa', err);
    throw err.name;
  }
  finally {
    if (client) {
      client.close();
    }
  }
};


const stream = require('stream');

export const uploadStreamFile = async() => {
  try{
    const pass = new stream.PassThrough();
    const upload = new Upload({
      client: s3Client,
      params: {
        Bucket: 'test-bucket',
        Key: 'extracted-data/benda_mongo.json',
        Body: pass,
      },
    });
    await upload.done();
    return pass;
  }
  catch(err){
    log.error('pawoooooo', err);
    return;
  }
};

错误 #5:

TypeError: dest.on is not a functionat Cursor.pipe (_stream_readable.js:680:8)

Answer 1

查看错误堆栈跟踪后，问题可能与 MongoDB 驱动程序 provides a cursor in object mode 相关。而 Body Upload的参数需要传统流，适合Buffer 处理在这种情况下。
以您的原始代码为引用，您可以尝试提供 Transform处理这两个要求的流。
请考虑以下代码:

import { Transform } from 'stream';
import { MongoClient } from 'mongodb';
import { S3Client } from '@aws-sdk/client-s3';
import { Upload } from '@aws-sdk/lib-storage';
const s3Client = new S3Client({ region: env.AWS_REGION });

export const uploadMongoStreamToS3 = async (connectionString, collectionName) => {
  let client;

  try {
    client = await MongoClient.connect(connectionString);
    const db = client.db();
    const readStream = db.collection(collectionName).find('{}').limit(5).stream();
    // We are creating here a Transform to adapt both sides
    const toJSONTransform = new Transform({
      writableObjectMode: true,
      transform(chunk, encoding, callback) {
        this.push(JSON.stringify(chunk) + '\n');
        callback();  
      }  
    });

    readStream.pipe(toJSONTransform);
 
    const upload = new Upload({
      client: s3Client,
      params: {
        Bucket: 'test-bucket',
        Key: 'extracted-data/benda_mongo.json',
        Body: toJSONTransform,
      },
    });
    
    await upload.done(); 
  }
  catch (err) {
    log.error(err);
    throw err.name;
  }
  finally {
    if (client) {
      client.close();
    }
  }

};

在代码中， toJSONTransform我们将流的可写部分定义为对象模式；相比之下，可读部分将适合从 S3 Upload 中读取。方法……至少，我希望如此。
关于您报告的第二个错误，与 dest.on 相关的错误。 ，我最初认为，我给你写了关于这种可能性的错误，因为在 uploadStreamFile您正在返回 Promise ，而不是流，而您正在传递 Promise到 pipe方法，它需要一个流，基本上是你返回了错误的变量。但我没有意识到您正在尝试通过 PassThrough流作为 Upload 的参数方法:请注意，这个流不包含任何信息，因为您没有向它传递任何信息，从 MongoDB 查询获得的可读流的内容永远不会传递给回调，也不会传递给 Upload本身。