python - 这是将某种类型的所有实体导出到 csv 文件的好方法吗？

Question

我有数百万个特定类型的实体，我想将其导出到 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)

Answer 1

当前解决方案的问题是，每次写入 blob 存储区时，内存都会增加。 blobstore 是不可变的，并且会立即从内存中写入所有数据。

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