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

Posted 2023-02-24

【中文标题】使用 Django REST 框架进行批量插入的最佳设计模式是啥？

【英文标题】：What is the best design pattern for batch insertion using the Django REST Framework?使用 Django REST 框架进行批量插入的最佳设计模式是什么？

【发布时间】：2016-04-28 02:17:21

【问题描述】：

背景

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

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

问题

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

建议的方法

我正在考虑用 Celery 做这个，但我以前没有用过。我正在考虑在我现有的 DRF ModelViewSets 中覆盖 perform_create()（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 视图和序列化程序：

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 客户端排队他们的数据。如果它不合适，我不会致力于芹菜。这是最好的解决方案，是否存在明显的问题，还是有更好的替代方案？

【问题讨论】：

用序列化器的异步create()方法代替视图会不会更好，这样你也可以在排队之前进行验证？

【参考方案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（再次强调，喜欢它，但它使测试变得有点棘手，并且有时会根据工作负载、配置等情况变得有点棘手） 它处理批量创建，因此如果您在很短的时间跨度（几秒或不到一秒）内提交了数千本书，则只有少数插入会在数据库上执行（而不是数千次插入）

缺点

您自己处理低级序列化，而不是 Celery “神奇地”完成它 您需要自己管理工作脚本（对其进行守护，可能将其打包为管理命令，处理重启等），而不是将其交给 Celery

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

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

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

【讨论】：

有用的答案 - 非常感谢。在挑选一个被接受之前，我会看看还会出现什么。非常感谢！