AppEngine：带有管道的实体架构迁移

Question

我很想知道在 Google App Engine 中迁移实体架构的最佳做法是什么。我们经常使用管道，我倾向于构建一个管道任务来处理这种迁移。所以这就是我想出的（在这个例子中，如果用户的年龄是质数，我存储）：

class MigrateUsers(pipeline.Pipeline):
  def run(self, keys):
    futures = []
    users = ndb.get_multi(keys)
    for user in users:
      user.age_is_prime = is_prime(user.age)
      futures.append(user.put_async())
    ndb.Future.wait_all(futures)

class Migration(pipeline.Pipeline):
  def run(self):
    all_results = []

    q = ds.User.query().filter()
    more = True
    next_cursor = None

    # Fetch user keys in batch and create MigrateUsers jobs
    while more:
      user_keys, next_cursor, more = \
        q.fetch_page(500, keys_only=True, start_cursor=next_cursor)
      all_results.append((yield MigrateUsers(keys=user_keys)))

    # Wait for them all to finish
    pipeline.After(*all_results)

我的问题是，我这样做对吗？我的“迁移”任务迭代所有用户以创建分段任务，这感觉有点笨拙。我确实看了一下mapreduce，但我觉得它不合适。我会很感激任何建议，如果您正在使用 mapreduce 并且不介意转换我的示例，我将非常感激。

Answer 1

MapReduce 非常适合迁移。根据我自己的经验，迁移通常意味着我需要检查所有实体，更新它们，然后将它们写回数据存储区。在这种情况下，我真的只需要“map”部分，不需要mapreduce的“reduce”部分。

使用 mapreduce 的好处是它会自动在不同实例上并行批处理您的实体，因此您的操作将比在管道示例中串行运行快得多。 MR SDK 有一个 DatastoreInputReader()，它将获取给定类型的每个实体，并在每个实体上调用一个映射函数，您只需提供该映射函数：

from mapreduce import operation as op
def prime_age_map(user_entity):
    user_entity.age_is_prime = is_prime(user.age)
    if user_entity.age_is_prime:
        yield op.db.Put(user_entity)

这里有一些样板代码我没有包括在内，因为我还没有切换到最新的 SDK，我所拥有的可能是不正确的，但它应该很简单，因为你只使用了一半的管道。

我不确定你的例子有多真实，但如果它是真实的并且你有很多实体，那么预先计算主要值会更好（http://primes.utm.edu/lists/small/1000.txt - 只有前 30 个左右是合理的年龄值)，并对这些年龄值执行特定查询并更新这些实体，而不是遍历整个 Kind。您可以使用 MapReduce 管道执行此操作，但您必须修改给定的 DatastoreInputReader 以发出比获取整个 Kind 更具体的查询。

Answer 2

我强烈建议您查看应用引擎 TaskQueues 以进行架构迁移。它比后端或 MapReduce、IMO 更容易设置和操作。你可以在这里找到一些信息：blog entry。