Nodejs,Cloud Firestore上传任务 - 验证错误:错误:套接字挂起

Question

我正在编写一个运行API调用的函数,并通过偏移按顺序从一个巨大的数据库中请求JSON.解析JSON响应,然后将其中的后续数据上载到我们的Cloud Firestore服务器.

Nodejs(节点6.11.3)和最新的Firebase Admin SDK

信息按预期进行解析,并完美地打印到控制台.但是,当数据尝试上传到我们的Firestore数据库时,控制台会发送垃圾邮件并显示错误消息:

验证错误:错误:套接字挂起

(node:846)UnhandledPromiseRejectionWarning:未处理的promise promise(拒绝ID:-Number-):错误:从插件获取元数据失败,错误:socket挂断

偶尔:

验证错误:错误:读取ECONNRESET

forEach函数从下载的JSON中收集项目,并在上载到Firestore数据库之前处理数据.每个JSON最多有1000项数据(1000个文档值)通过forEach函数.我知道如果函数在上传集完成之前重复,这可能是个问题吗？

我是一个编码新手,并且明白这个功能的控制流程并不是最好的.但是,我找不到有关控制台打印的错误的任何信息.我可以找到有关套接字挂起的大量信息,但在Auth错误部分没有.

我使用生成的服务帐户JSON作为访问我们的数据库的凭据,该数据库使用firebase-adminsdk帐户.我们的数据库读/写规则目前是开放的,允许任何访问(因为我们正在开发没有真正的用户).

这是我的功能:

Firebase初始化和偏移归零

 const admin = require('firebase-admin');
    var serviceAccount = require("JSON");
    admin.initializeApp({
    credential: admin.credential.cert(serviceAccount),
    databaseURL: "URL"
    });
    var db = admin.firestore();
    var offset = 0;
    var failed = false;

运行该功能并设置HTTP标头

var runFunction = function runFunction() {
    var https = require('https');
    var options = {
        host: 'website.com',
        path: (path including an offset and 1000 row specifier),
        method: 'GET',
        json: true,
        headers: {
            'content-type': 'application/json',
            'Authorization': 'Basic ' + new Buffer('username' + ':' + 'password').toString('base64')
        }
    };

如果我们尚未到达API的响应结束,则运行HTTP请求并重新运行该功能

if (failed === false) {
        var req = https.request(options, function (res) {
            var body = '';
            res.setEncoding('utf8');
            res.on('data', function (chunk) {
                body += chunk;
            });
            res.on('end', () => {
                console.log('Successfully processed HTTPS response');
                body = JSON.parse(body);
                if (body.hasOwnProperty('errors')) {
                    console.log('Body ->' + body)
                    console.log('API Call failed due to server error')
                    console.log('Function failed at ' + offset)
                    req.end();
                    return
                } else {
                    if (body.hasOwnProperty('result')) {
                        let result = body.result;
                        if (Object.keys(result).length === 0) {
                            console.log('Function has completed');
                            failed = true;
                            return;
                        } else {
                            result.forEach(function (item) {
                                var docRef = db.collection('collection').doc(name);
                                console.log(name);
                                var upload = docRef.set({
                                    thing: data,
                                    thing2: data,
                                })
                            });
                            console.log('Finished offset ' + offset)
                            offset = offset + 1000;
                            failed = false;
                        }
                        if (failed === false) {
                            console.log('Function will repeat with new offset');
                            console.log('offset = ' + offset);
                            req.end();
                            runFunction();
                        } else {
                            console.log('Function will terminate');
                        }
                    }
                }
            });
        });
        req.on('error', (err) => {
            console.log('Error -> ' + err)
            console.log('Function failed at ' + offset)
            console.log('Repeat from the given offset value or diagnose further')
            req.end();
        });
        req.end();
    } else {
        req.end();
    }
    };
    runFunction();

任何帮助将不胜感激!

UPDATE

我刚刚尝试更改我一次拉出的JSON行,然后使用该函数一次上传 - 从1000下降到100.套接字挂起错误的频率较低,因此肯定是由于数据库超载.

理想情况下,如果每个forEach数组迭代在开始之前等待上一次迭代完成,那将是完美的.

更新#2

我已经安装了异步模块,我目前正在使用async.eachSeries函数一次执行一个文档上传.上传中期的所有错误都消失了 - 但是这个功能需要花费大量的时间才能完成(158,000个文档大约需要9个小时).我更新的循环代码是这样的,实现了一个计数器:

async.eachSeries(result, function (item, callback) {
    // result.forEach(function (item) {
    var docRef = db.collection('collection').doc(name);
    console.log(name);
    var upload = docRef.set({
      thing: data,
      thing2: data,
    }, { merge: true }).then(ref => {
        counter = counter + 1
        if (counter == result.length) {
            console.log('Finished offset ' + offset)
            offset = offset + 1000;
            console.log('Function will repeat with new offset')
            console.log('offset = ' + offset);
            failed = false;
            counter = 0
            req.end();
            runFunction();
        }
        callback()
    });
});

此外,一段时间后,数据库返回此错误:

(node:16168)UnhandledPromiseRejectionWarning:未处理的promise promise(拒绝ID:-Number-):错误:数据存储区操作超时,或者数据暂时不可用.

好像现在我的功能花了太长时间......而不是不够长.有没有人有任何关于如何在没有明确错误的情况下更快地运行的建议？

Answer 1

作为此循环一部分的写入请求只是超出了 Firestore 的配额 - 因此服务器拒绝了其中的大部分。

为了解决这个问题，我将我的请求转换为一次上传 50 个左右项目的块，并使用 Promises 确认何时进行下一个块上传。

答案发布在这里 ->在 Node.js 中一次迭代 50 个项目的数组，我的工作代码的模板如下：

async function uploadData(dataArray) {
  try {
    const chunks = chunkArray(dataArray, 50);
    for (const [index, chunk] of chunks.entries()) {
      console.log(` --- Uploading ${index + 1} chunk started ---`);
      await uploadDataChunk(chunk);
      console.log(`---Uploading ${index + 1} chunk finished ---`);
    }
  } catch (error) {
    console.log(error)
    // Catch en error here
  }
}

function uploadDataChunk(chunk) {
  return Promise.all(
    chunk.map((item) => new Promise((resolve, reject) => {
      setTimeout(
        () => {
          console.log(`Chunk item ${item} uploaded`);
          resolve();
        },
        Math.floor(Math.random() * 500)
      );
    }))
  );
}

function chunkArray(array, chunkSize) {
  return Array.from(
    { length: Math.ceil(array.length / chunkSize) },
    (_, index) => array.slice(index * chunkSize, (index + 1) * chunkSize)
  );
}

将数据数组传递给 uploadData - 使用 uploadData(data); 并将每个项目的上传代码发布到 chunk.map 函数内 setTimeout 块内的 uploadDataChunk（resolve() 行之前）。