1

我有数百万个特定类型的实体，我想导出到 csv 文件。以下代码将实体以 1000 个为单位批量写入 blob，同时保持 blob 打开并将下一批实体推迟到任务队列。当没有更多实体要获取时，blob 就完成了。这似乎适用于我的大部分本地测试，但我想知道：

如果我在我的生产数据上运行它之前错过了任何陷阱或极端情况，并为数据存储读取产生了 $s。

如果超过最后期限或在批处理写入 blob 时内存耗尽，则此代码默认为当前批处理的开始以再次运行任务，这可能会导致大量重复。有什么建议可以解决这个问题吗？

def entities_to_csv(entity_type,blob_file_name='',cursor='',batch_size=1000):
  more = True
  next_curs = None
  q = entity_type.query()
  results,next_curs,more = q.fetch_page(batch_size,start_cursor=Cursor.from_websafe_string(cursor))
  if results:
    try:
      if not blob_file_name:
        blob_file_name = files.blobstore.create(mime_type='text/csv',_blob_uploaded_filename='%s.csv' % entity_type.__name__)

      rows = [e.to_dict() for e in results]
      with files.open(blob_file_name, 'a') as f:
        writer = csv.DictWriter(f,restval='',extrasaction='ignore',fieldnames=results[0].keys())
        writer.writerows(rows)

      if more:
        deferred.defer(entity_type,blob_file_name,next_curs.to_websafe_string())
      else:
        files.finalize(blob_file_name)
    except DeadlineExceededError:
      deferred.defer(entity_type,blob_file_name,cursor)

稍后在代码中，类似于：

deferred.defer(entities_to_csv,Song)
4

2 回答 2

1

当前解决方案的问题在于，每次写入到 blobstore 时，您的内存都会增加。blobstore 是不可变的，可以一次从内存中写入所有数据。

您需要在可以将所有记录保存在内存中的后端上运行该作业，您需要在应用程序中定义一个后端并使用 defer 调用_target='<backend name>'。

于 2012-07-11T14:44:09.657 回答
0

看看这个 Google I/O 视频，几乎描述了你想用 MapReduce 做什么，从视频中的 23:15 开始。你要的代码是27:19

https://developers.google.com/events/io/sessions/gooio2012/307/

于 2012-07-11T15:14:16.860 回答