CSV到Cloudant导入大数据集

Question

我正在使用基于Nodejs的脚本将大型csv文件导入ibm cloudant数据库。我对该脚本的处理方式如下：

• 脚本1：使用FTP连接从远程服务器下载ZIP文件。
• Script1：遍历下载的zip文件并从中提取CSV文件。
• 脚本1：使用文件读取和写入流将每个CSV文件上传到AWS S3存储桶。
• Script2：连接到S3存储桶并遍历在特定存储桶位置在该位置找到的每个CSV文件。
• Script2：使用流读取CSV文件，并通过管道将其转换为JSON格式的CSV数据。
• Script2：连接到IBM cloudant数据库。
• Script2：通过发出GET请求，对每个JSON对象逐一检查现有cloudant文档。
• Script2：如果找到了文档，则使用JSON对象进行更新，否则将作为新文档创建cloudant条目，并具有单独的API请求。
• Script2：如果所有CSV数据都已成功迁移到cloudant数据库，则从S3存储桶中删除CSV文件。

到目前为止有效的方法：

• 从FTP下载ZIP文件，然后从ZIP提取CSV文件并将其上传到S3存储桶。
• [如果我们只是在终端上打印每个转换的JSON对象，则使用npm包“ csvtojson”将CSV文件作为流读取并将其转换为JSON对象，就像魅力一样工作。
• 成功连接到IBM cloudant数据库。

面临的问题：

• 在尝试检查每个JSON对象在cloudant处的现有记录的循环中，它只是越过很少的记录，并最终导致与内存/堆相关的错误，而且它存在cloudant速率限制的问题，因为随着查找并在循环内进行导入操作。

确切错误：

• 致命错误：无效的标记压缩接近堆限制分配失败-JavaScript堆内存不足。
• IBM Cloudant上的请求太多。

注意：

• FTP服务器上可能有多个ZIP文件。
• ZIP文件中可能有多个CSV文件。
• 我一直尝试导入的CSV文件之一包含约130万条记录，文件大小约为300MB。
• 我没有选择考虑在IBM Cloudant上提高此要求的速率限制。

有人可以提出建议的最佳方法来解决此要求，而不会遇到如上所述的问题吗？也欢迎任何其他可解决此要求的npm软件包。也欢迎任何关于改变我的开发方法的建议。

Answer 1

假设您的CSV行按ID排序，则可以先进行批量获取：

请求：

POST /db/_all_docs HTTP/1.1
Accept: application/json
Content-Length: 70
Content-Type: application/json
Host: localhost:5984

{
    "keys" : [
        "a_id123",
        "b_id123",
        "c_id123",
        "d_id123",
    ]
}

然后您将获得以下内容：响应

{
    "total_rows" : 2666,
    "rows" : [
        {
            "value" : {
                "rev" : "1-a3544d296de19e6f5b932ea77d886942"
            },
            "id" : "a_id123",
            "key" : "a_id123"
        },
        {
            "value" : {
                "rev" : "1-91635098bfe7d40197a1b98d7ee085fc"
            },
            "id" : "b_id123",
            "key" : "b_id123"
        }
    ],
    "offset" : 0
}

现在您可以遍历响应，查看哪些文档存在而哪些不存在。

此后，您可以批量添加不存在的文档：

请求：

POST /db/_bulk_docs HTTP/1.1
Accept: application/json
Content-Length: 109
Content-Type:application/json
Host: localhost:5984

{
    "docs": [
        {
            "_id": "c_id123",
            "foo": "bar",
        },
        {
            "_id": "d_id123",
            "foo": "bar c",
        },
        {
            "_id": "a_id123",
            "_rev": "1-a3544d296de19e6f5b932ea77d886942",
            "foo": "updated",
        }
    ]
}

批量获取文件：https://docs.couchdb.org/en/stable/api/database/bulk-api.html#post--db-_all_docs

用于批量创建/更新的文档：https://docs.couchdb.org/en/stable/api/database/bulk-api.html#db-bulk-docs