python - 如何批量上传数据到appengine datastore？旧方法不起作用

Question

这应该是一个相当普遍的要求，也是一个简单的过程:将数据批量上传到应用引擎数据存储区。

但是，stackoverflow(下面的链接*)中提到的旧解决方案似乎都不再有效。 bulkloader 方法是使用 DB API 上传到数据存储时最合理的解决方案，但不适用于 NDB API

现在 bulkloader 方法似乎已被弃用，文档中仍然存在的旧链接会导致错误的页面。这是一个例子

https://developers.google.com/appengine/docs/python/tools/uploadingdata

此页面上仍然存在上述链接:https://developers.google.com/appengine/docs/python/tools/uploadinganapp

现在推荐的批量加载数据的方法是什么？

两个可行的替代方案似乎是 1) 使用 remote_api 或 2) 将 CSV 文件写入 GCS 存储桶并从中读取。有人有成功使用这两种方法的经验吗？

任何指点将不胜感激。谢谢!

[*以下链接提供的解决方案不再有效]

[1] how does one upload data in bulk to a google appengine datastore?

[2] How to insert bulk data in Google App Engine Datastore?

Answer 1

方法一:使用remote_api

如何:编写一个 bulkloader.yaml 文件并从终端使用“appcfg.py upload_data”命令直接运行它我不推荐这种方法有几个原因:1. 巨大的延迟 2. 不支持 NDB

方法二:GCS并使用mapreduce

上传数据文件到 GCS:

使用“storage-file-transfer-json-python ” github 项目 (chunked_transfer.py) 将文件从本地系统上传到 gcs。确保从应用引擎管理控制台生成正确的“client-secrets.json”文件。

Mapreduce:

使用“appengine-mapreduce”github 项目。将“mapreduce”文件夹复制到您的项目顶级文件夹。

将以下行添加到您的 app.yaml 文件中:

includes:
  - mapreduce/include.yaml

下面是你的main.py文件

import cgi
import webapp2
import logging
import os, csv
from models import DataStoreModel
import StringIO
from google.appengine.api import app_identity
from mapreduce import base_handler
from mapreduce import mapreduce_pipeline
from mapreduce import operation as op
from mapreduce.input_readers import InputReader

def testmapperFunc(newRequest):
    f = StringIO.StringIO(newRequest)
    reader = csv.reader(f, delimiter=',')
    for row in reader:
        newEntry = DataStoreModel(attr1=row[0], link=row[1])
        yield op.db.Put(newEntry)

class TestGCSReaderPipeline(base_handler.PipelineBase):
    def run(self, filename):
        yield mapreduce_pipeline.MapreducePipeline(
                "test_gcs",
                "testgcs.testmapperFunc",
                "mapreduce.input_readers.FileInputReader",
                mapper_params={
                    "files": [filename],
                    "format": 'lines'
                },
                shards=1)

class tempTestRequestGCSUpload(webapp2.RequestHandler):
    def get(self):
        bucket_name = os.environ.get('BUCKET_NAME',
                                     app_identity.get_default_gcs_bucket_name())

        bucket = '/gs/' + bucket_name
        filename = bucket + '/' + 'tempfile.csv'

        pipeline = TestGCSReaderPipeline(filename)
        pipeline.with_params(target="mapreducetestmodtest")
        pipeline.start()
        self.response.out.write('done')

application = webapp2.WSGIApplication([
    ('/gcsupload', tempTestRequestGCSUpload),
], debug=True)

要记住:

1. Mapreduce 项目使用现已弃用的“Google Cloud Storage Files API”。因此无法保证 future 的支持。
2. Map reduce 会增加数据存储读写的少量开销。

方法三:GCS和GCS客户端库

1. 使用上述文件传输方法将 csv/text 文件上传到 gcs。
2. 使用 gcs 客户端库(将“cloudstorage”文件夹复制到您的应用程序顶级文件夹)。

将以下代码添加到应用程序 main.py 文件中。

import cgi
import webapp2
import logging
import jinja2
import os, csv
import cloudstorage as gcs
from google.appengine.ext import ndb
from google.appengine.api import app_identity
from models import DataStoreModel

class UploadGCSData(webapp2.RequestHandler):
    def get(self):
        bucket_name = os.environ.get('BUCKET_NAME',
                                     app_identity.get_default_gcs_bucket_name())
        bucket = '/' + bucket_name
        filename = bucket + '/tempfile.csv'
        self.upload_file(filename)

    def upload_file(self, filename):
        gcs_file = gcs.open(filename)
        datareader = csv.reader(gcs_file)
        count = 0
        entities = []
        for row in datareader:
            count += 1
                newProd = DataStoreModel(attr1=row[0], link=row[1])
                entities.append(newProd)

            if count%50==0 and entities:
                ndb.put_multi(entities)
                entities=[]

        if entities:
            ndb.put_multi(entities)

application = webapp2.WSGIApplication([
    ('/gcsupload', UploadGCSData),
], debug=True)