并行代码执行 python2.7 ndb答案

【问题标题】：parallel code execution python2.7 ndb并行代码执行 python2.7 ndb
【发布时间】：2012-03-28 20:39:34
【问题描述】：

在我的应用程序中，对于一个处理程序，我需要获取一堆实体并为每个实体执行一个函数。

我拥有我需要的所有实体的钥匙。获取它们后，我需要为它们中的每一个执行 1 或 2 个实例方法，这会大大降低我的应用程序的速度。对 100 个实体执行此操作大约需要 10 秒，这太慢了。

我试图找到一种方法来获取实体并并行执行这些功能以节省时间，但我不确定哪种方法最好。

我尝试了 _post_get_hook，但我有一个未来的对象，需要调用 get_result() 并执行钩子中的函数，这在 sdk 中工作正常，但在调用 Python 时得到了很多“最大递归深度” objec'，但我真的不明白为什么，错误信息也不是很详细。

我在搜索什么 Pipeline api 或 ndb.Tasklets？

atm 我正在反复试验，但如果有人能引导我走向正确的方向，我会很高兴。

编辑

我的代码类似于文件系统，每个文件夹都包含其他文件夹和文件。集合的路径设置在另一个实体上，因此要序列化集合实体，我需要获取引用的实体并获取路径。在 Collection 上， serialized_assets() 函数包含的实体越多，它的速度就越慢。如果我可以并排为每个包含的资产执行序列化函数，它会加快速度。

class Index(ndb.Model):
    path = ndb.StringProperty()


class Folder(ndb.Model):
    label = ndb.StringProperty()
    index = ndb.KeyProperty()

    # contents is a list of keys of contaied Folders and Files
    contents = ndb.StringProperty(repeated=True)    

    def serialized_assets(self):
        assets = ndb.get_multi(self.contents)

        serialized_assets = []
        for a in assets:
            kind = a._get_kind()
            assetdict = a.to_dict()
            if kind == 'Collection':
                assetdict['path'] = asset.path
                # other operations ...
            elif kind == 'File':
                assetdict['another_prop'] = asset.another_property
                # ...
            serialized_assets.append(assetdict)

        return serialized_assets

    @property
    def path(self):
        return self.index.get().path


class File(ndb.Model):
    filename = ndb.StringProperty()
    # other properties....

    @property
    def another_property(self):
        # compute something here
        return computed_property

EDIT2：

    @ndb.tasklet
    def serialized_assets(self, keys=None):
        assets = yield ndb.get_multi_async(keys)
        raise ndb.Return([asset.serialized for asset in assets])

这个tasklet代码可以吗？

【问题讨论】：

函数慢是因为它们进行大量计算，还是因为它们等待 RPC？如果前者，线程或任务队列是你最好的选择；如果是后者，NDB 小任务就是你想要的。详细说明，我可以发布一个答案来展示其中一个。
@NickJohnson 主要是在等待 RPC。我用更多细节编辑了这个问题。

标签： google-app-engine python-2.7 app-engine-ndb

【解决方案1】：

由于您的函数的大部分执行时间都花在等待 RPC 上，因此 NDB 的 async 和 tasklet 支持是您最好的选择。这在here 中有详细描述。根据您的要求，最简单的用法可能是使用 ndb.map 函数，如下所示（来自文档）：

@ndb.tasklet
def callback(msg):
  acct = yield ndb.get_async(msg.author)
  raise tasklet.Return('On %s, %s wrote:\n%s' % (msg.when, acct.nick(), msg.body))

qry = Messages.query().order(-Message.when)
outputs = qry.map(callback, limit=20)
for output in outputs:
  print output

为查询返回的每个实体调用回调函数，它可以执行它需要的任何操作（使用_async 方法和yield 异步执行它们），完成后返回结果。因为回调是一个tasklet，并且使用yield来进行异步调用，NDB可以并行运行它的多个实例，甚至可以批处理一些操作。

【讨论】：

我不认为我可以使用 map，因为我在我已经知道的键列表上使用 ndb.get_multi()（或者至少我没有看到任何关于将 map 与 ndb 一起使用的文档。 get_multi())。
我认为有一些拼写错误或者可能是旧文档：ndb.get_async(msg.author) 应该是 msg.author.get_async() 并且 tasket.Return 是 ndb.Return ...？！
acct 是一个 Future 对象，因此尝试在 Return 中格式化字符串会引发 AttributeError: 'Future' object has no attribute 'nick'
如果您使用 ndb.get_multi 获取一组引用，您可以在 tasklet 中执行单独的异步获取。 NDB 将根据需要对它们进行批量处理。您对第二个问题是正确的；第三，acct 不是未来对象 - 在 tasklet 中产生未来会返回对象本身。
我在我的问题中添加了一个edit2。我现在有一个更快的 tasklet，但大约 400 个资产仍然需要大约 6 秒。代码可以吗？

【解决方案2】：

管道 API 对于您想要做的事情来说太过分了。有什么理由不能只使用任务队列吗？

使用初始请求获取所有实体键，然后为每个键排队一个任务，让该任务执行每个实体的 2 个功能。并发将基于为该任务队列配置的并发请求数。

【讨论】：