导入时 GCP 身份验证超时

Question

我正在使用 GCP CLI 中的简单节点脚本将 16,000 条记录从 CSV 导入 Firestore。

CSV 有四列区域，每列都写为集合中的一个新文档。我必须异步处理 CSV，因为从 CSV 写入的每一行都链接到前一行中的数据。因此，导入需要 5 个多小时。

该过程不断失败，并出现以下错误： 错误：16 UNAUTHENTICATED：请求具有无效的身份验证凭据。预期的 OAuth 2 访问令牌、登录 cookie 或其他有效的身份验证凭据。

我怀疑这是因为这是一个很长的异步任务，因此身份验证令牌即将到期。从我所读到的身份验证令牌的寿命为 1 小时。

我在下面简化了我的代码来模拟 CVS 导入问题。在 Google Cloud Platform 命令行中运行时，这会重现错误。

const {Firestore} = require('@google-cloud/firestore');
const db = new Firestore();
var previous = '';

async function processRecord(record) {
    console.log(record);
    console.log(record);
    let collectionRef = db.collection('groups');
    const snapshot = await collectionRef.where('name', '==', 'World').get();
    let data = {
        name: record,
        previous: previous,
        created: Firestore.Timestamp.now()
    }
    let docRef = await db.collection('testing').add(data);
    previous = docRef.id;
}

async function importCsv(csvFileName) {
    var records = [];
    for(var i = 0; i < 16000; i++) {
        records.push(`Filler: ${i}`);
    }

    // const fileContents = await readFile(csvFileName, 'utf8');
    // const records = await parse(fileContents, { columns: true });

    for (const record of records) {
        await processRecord(record);
    }
    console.log(`Processed ${records.length} records`);
}

importCsv(process.argv[2]).catch(e => console.error(e));

我可以做些什么来完成我的导入？

Answer 1

已编辑：在给出更多细节后修改了整个答案（代码示例和更好的要求描述）

避免由于身份验证令牌过期而导致的错误的最佳方法是加快加载速度（16k 的 5 小时时间太长了）。这可以通过多种方式实现，但在查看了提供的代码后，我创建了两个 nodejs 脚本，它们可以满足您创建“反向链表”的需求，而无需花费超过 1 分钟的时间来加载整个 16k 记录集。

解释：创建记录并在同一个事务中读取它是一种缓慢的方法，最好先创建记录，然后查询完整的数据集并相应地更新。此外，强烈建议使用批处理进行写入和更新操作，因为服务器端库会并行化单个写入，有关此的更多信息可以在下一个链接中找到：Batched writes:

对于批量数据输入，请使用具有并行单独写入的服务器客户端库。批量写入的性能优于序列化写入，但不优于并行写入。您应该使用服务器客户端库进行批量数据操作，而不是移动/网络 SDK。

脚本： 第一部分是加载整个数据集，我使用代码中提供的虚拟负载来举例说明这一点，但是您可以用 csv 中加载的行替换记录变量文件。另请注意，我使用数字索引填充 previous 字段，这对于跟踪我们加载记录的顺序很重要。

batch_writing.js

const {Firestore} = require('@google-cloud/firestore');
const firestore = new Firestore();
async function batch_writting() {
    var records = [];
    let writeBatch = firestore.batch();

    for(var i = 0; i < 17000; i++) {
        records.push(`Filler: ${i}`);
    }
    let index = 0;
    for (const record of records) {
        let documentRef = firestore.collection('testing').doc();
        let data = {
            name: record,
            previous: index,
            created: Firestore.Timestamp.now()
        }
        writeBatch.create(documentRef,data);
        if((index+1) % 500 === 0){
            writeBatch.commit().then(() => {
                console.log('Successfully executed batch.');
            }).catch(e => console.error(e));
            writeBatch = firestore.batch();
        }
        index++;
    }
    writeBatch.commit().then(() => {
        console.log('Successfully executed batch.');
    }).catch(e => console.error(e));
}

batch_writting().catch(e => console.error(e));

编写部分完成后，previous 字段填充了递增索引，然后我们可以继续查询最近创建的完整数据集并继续使用正确的文档 ID 更新 previous 键 从上一行加载。这是通过在增量索引中填充的previous 字段对查询进行排序来完成的。请注意，第一条记录的数值为 0，因为它没有先前加载的行。

update_writing.js

const {Firestore} = require('@google-cloud/firestore');
const firestore = new Firestore();
async function batch_update(){
    let query = firestore.collection('testing');
    query.orderBy('previous', 'asc').get().then(querySnapshot => {
        let writeBatch = firestore.batch();
        let previousID = 0;
        let index = 0;
        querySnapshot.forEach(documentSnapshot => {
            writeBatch.update(documentSnapshot.ref,{previous: previousID})
            previousID = documentSnapshot.id;
            if((index+1) % 500 === 0){
                writeBatch.commit().then(() => {
                    console.log('Successfully executed batch.');
                }).catch(e => console.error(e));
                writeBatch = firestore.batch();
            }
            index++;
        });
        writeBatch.commit().then(() => {
            console.log('Successfully executed batch.');
        }).catch(e => console.error(e));
    });
}

batch_update().catch(e => console.error(e));

如果您需要查询更多字段或其他集合，请随时更改脚本，但请记住避免查询特定字段并支持收集完整数据集的查询。此外，如果您需要额外的写入，请使用批量写入。