json - Nodejs、Cloud Firestore 上传任务 - 身份验证错误 :Error: socket hang up

Question

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

Nodejs (Node 6.11.3) 和最新的 Firebase Admin SDK

信息按预期解析，并完美打印到控制台。但是，当数据尝试上传到我们的 Firestore 数据库时，控制台会收到错误消息:

Auth error:Error: socket hang up

(node:846) UnhandledPromiseRejectionWarning: Unhandled promise rejection(rejection id: -Number-): Error: Getting metadata from plugin failed witherror: socket hang up

偶尔:

Auth error:Error: read 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 header

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();

如有任何帮助，我们将不胜感激!

更新

我刚刚尝试更改我一次提取的 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: Unhandled promise rejection (rejection id: -Number-): Error: The datastore operation timed out, or the data was temporarily unavailable.

似乎现在我的函数花费的时间太长了……而不是不够长。有没有人对如何在没有规定错误的情况下更快地运行有任何建议？

Answer 1

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

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

答案张贴在这里 -> Iterate through an array in blocks of 50 items at a time in node.js ，我的工作代码模板如下:

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 block 内的 uploadDataChunk 中(在 resolve() 行之前)。