django - 使用 Django REST 框架进行批量插入的最佳设计模式是什么？

Question

背景

我有一个 Django 应用程序，它允许通过 Django REST 框架插入记录。

查询电子表格和其他数据库的客户端应用程序将定期逐行批量插入记录。 REST API 允许从 Django 中抽象出这些处理数据转换等的其他应用程序。

问题

我想将实际的记录插入与 API 分离，以提高容错性和可扩展性的潜力。

建议的方法

我正在考虑用 celery 做这个，但我以前没有用过。我正在考虑覆盖 perform_create()在我现有的 DRF 模型 View 集中( perform_create() 被添加到 DRF 3.0 中)来创建工作人员将在后台抓取和处理的 Celery 任务。

DRF 文档说 perform_create()应该“应该通过调用 serializer.save() 来保存对象实例”。我想知道，在我的情况下，我是否可以忽略此建议，而是让我的 Celery 任务调用适当的序列化程序来执行对象保存。

例子

例如，如果我有几个模型:

class Book(models.Model):
    name = models.CharField(max_length=32)

class Author(models.Model):
    surname = models.CharField(max_length=32)

我有这些模型的 DRF View 和序列化程序:

class BookSerializer(serializers.ModelSerializer):
    class Meta:
        model = Book

class AuthorSerializer(serializers.ModelSerializer):
    class Meta:
        model = Author

class BookViewSet(viewsets.ModelViewSet):
    queryset = Book.objects.all()
    serializer_class = Book

class AuthorViewSet(viewsets.ModelViewSet):
    queryset = Author.objects.all()
    serializer_class = Author

在例如覆盖 perform_create() 是个好主意吗？ BookViewSet :

def perform_create(self, serializer):
    create_book_task(serializer.data)

哪里create_book_task分别是这样的:

@shared_task
def create_book_task(data):
    serializer = BookSerializer(data=data)
    serializer.save()

我真的没能找到任何其他开发人员做类似事情或试图解决同样问题的例子。我是否过于复杂了？当涉及到物理插入时，我的数据库仍然是限制因素，但至少它不会阻止 API 客户端排队他们的数据。如果 Celery 不合适，我不会致力于它。这是最好的解决方案，是否存在明显的问题，或者是否有更好的替代方案？

Answer 1

我发现您的方法是合理的，Celery 很棒，除了在我的经验中可能会变得有点讨厌的一些边界案例(但我不希望在您在问题中概述的用例中遇到这种情况)。

但是，请考虑以下使用 Redis 的简化方法。它有一些优点和缺点。

在 BookViewSet 中:

from redis import StrictRedis
from rest_framework import viewsets, renderers

redis_client = StrictRedis()

class BookViewSet(viewsets.ModelViewSet):
    queryset = Book.objects.all()
    serializer_class = Book

    def perform_create(self, serializer):
        json = renderers.JSONRenderer().render(serializer.data)
        redis_client.lpush('create_book_task', json)

在单独的工作脚本中:

from django.utils.six import BytesIO
from redis import StrictRedis
from rest_framework.parsers import JSONParser
from myproject import BookSerializer, Book

MAX_BATCH_SIZE = 1000

def create_book_task():
    bookset = []
    for json in redis_client.brpop(('create_book_task',)):
       stream = BytesIO(json)
       data = JSONParser().parse(stream)
       serializer = BookSerializer(data=data)
       assert serializer.is_valid()
       bookset.append(serializer.instance)
       if len(bookset) >= MAX_BATCH_SIZE:
           break

    if len(bookset) > 0:
        Book.objects.bulk_create(bookset)

while True:
    create_book_task()

优点

你不需要添加 Celery(同样，喜欢它，但它使测试变得有点棘手，有时会根据工作负载、配置等变得有点麻烦)

它处理批量创建，因此如果您在短时间内(几秒或不到一秒)提交了数千本书，则只会在 DB 上执行少量插入(而不是数千次插入)

缺点

您自己负责处理低级序列化，而不是 Celery“神奇地”执行此操作

您将需要自己管理工作脚本(对其进行守护，可能将其打包为管理命令，处理重启等)而不是将其交给 Celery

当然，以上是第一种方法，您可能希望使其更通用以重用其他模型，将 MAX_BATCH_SIZE 移至您的设置，使用酸洗代替 JSON 或根据您的具体情况进行各种其他调整、改进或设计决策需要。

最后，我可能会采用我的回答中概述的方法，除非您预计还有其他几项任务将被卸载到异步处理中，在这种情况下，使用 Celery 的情况会变得更加强大。

PS:由于实际插入将异步完成，请考虑使用 202 Accepted 进行响应response code而不是 201 Created (除非这搞砸了您的客户)。