通过 API 调用发送大型字典会中断开发服务器

Question

我正在运行一个带有 postgreSQL 数据库的 django 应用程序，我正在尝试向数据库发送一个非常大的字典（由时间序列数据组成）。

我的目标是尽快将我的数据写入数据库。我正在使用库请求通过 API 调用（使用 django REST 构建）发送数据：

我的 API 视图很简单：

@api_view(["POST"])
def CreateDummy(request):

    for elem, ts in request.data['time_series'] :
        TimeSeries.objects.create(data_json=ts)

    msg = {"detail": "Created successfully"}
    return Response(msg, status=status.HTTP_201_CREATED)

request.data['time_series'] 是一个巨大的字典，结构如下：

{Building1: {1:123, 2: 345, 4:567 .... 31536000: 2345}, .... Building30: {..... }}

这意味着我有 30 个键和 30 个值，而每个值都是一个具有 31536000 个元素的字典。

我的 API 请求如下所示（其中 data 是我上面描述的字典）：

 payload = {
            "time_series": data,
           } 

 requests.request(
        "post", url=endpoint, json=payload
    )

代码将时间序列数据保存到后端的 jsonb-field 中。现在，如果我只遍历字典的前 4 个元素，那就可以了。我可以在大约 1 分钟内获得这些数据。但是当我遍历整个字典时，我的开发服务器会关闭。我猜是因为内存不足。我得到一个requests.exceptions.ConnectionError: ('Connection aborted.', RemoteDisconnected('Remote end closed connection without response'))。整个字典是否在开始迭代之前保存到内存中？我对此表示怀疑，因为我在 python3 中读到，使用.items() 循环返回一个迭代器，这是执行此操作的首选方式。

有没有更好的方法来处理 django/python 中的大量 dicts？我应该循环遍历一半然后遍历另一半吗？还是有更快的方法？也许使用pandas？或者也许以不同的方式发送数据？我想我正在寻找最高效的方法来做到这一点。

如果需要，很乐意提供更多代码。

非常感谢任何帮助、提示或指南！提前致谢

EDIT2：我认为这不是我的 RAM 使用量或字典的大小。服务器关闭时，我还剩下 5GiB 的 RAM。 ~~而且dict的大小是1176bytes~~ dict要大很多，见cmets

EDIT3：我什至无法打印巨大的字典。然后它也会关闭

EDIT4：当拆分数据并发送它时，服务器不能一次处理它。但是当我尝试查询它时，服务器再次中断。它在我的生产服务器（nginx AWS RDS 设置）上中断，在我的本地开发服务器上中断。我很确定这是因为 django 无法使用我当前的设置处理这么大的查询。但是我该如何解决呢？

EDIT5：所以我正在寻找的是一个两部分的解决方案。一种用于创建数据，一种用于查询数据。我上面描述的数据的创建。但即使我将所有数据都输入数据库，我仍然会遇到问题。

我尝试通过创建数据而不是全部创建数据，而是每个时间序列都单独创建数据。所以让我们假设我的数据库中有这么大的数据，我尝试查询它。所有时间序列对象都属于一个网络，所以我这样尝试：

class TimeSeriesByTypeAndCreationMethod(ListAPIView):
    """Query time-series in specific network."""

    serializer_class = TimeSeriesSerializer

    def get_queryset(self):
        """Query time-series

        Query by name of network, type of data, creation method and
        source.
        """

        network = self.kwargs["name_network"]

        if TimeSeries.objects.filter(
            network_element__network__name=network,
        ).exists():
            time_series = TimeSeries.objects.filter(
                network_element__network__name=network,
            )
            return time_series
        else:
            raise NotFound()

但是查询像之前的数据创建一样中断了服务器。我认为这也是太多的数据负载。我想我可以使用原始 sql 避免破坏服务器......或者还有更好的方法吗？

EDIT6：相关模型：

class TimeSeries(models.Model):

    TYPE_DATA_CHOICES = [
        ....many choices...
    ]

    CREATION_METHOD_CHOICES = [
        ....many choices...
    ]

    description = models.CharField(
        max_length=120,
        null=True,
        blank=True,
    )

    network_element = models.ForeignKey(
        Building,
        on_delete=models.CASCADE,
        null=True,
        blank=True,
    )
    type_data = models.CharField(
        null=True,
        blank=True,
        max_length=30,
        choices=TYPE_DATA_CHOICES,
    )

    creation_method = models.CharField(
        null=True,
        blank=True,
        max_length=30,
        choices=CREATION_METHOD_CHOICES,
    )

    source = models.CharField(
        null=True,
        blank=True,
        max_length=300
    )

    data_json = JSONField(
        help_text="Data for time series in JSON format. Valid JSON expected."
    )

    creation_date = models.DateTimeField(auto_now=True, null=True, blank=True)

    def __str__(self):
        return f"{self.creation_method}:{self.type_data}"



class Building(models.Model):

    USAGE_CHOICES = [
        ...
    ]

    name = models.CharField(
        max_length=120,
        null=True,
        blank=True,
    )
    street = models.CharField(
        max_length=120,
        null=True,
        blank=True,
    )
    house_number = models.CharField(
        max_length=20,
        null=True,
        blank=True,
    )
    zip_code = models.CharField(
        max_length=5,
        null=True,
        blank=True,
    )
    city = models.CharField(
        max_length=120,
        null=True,
        blank=True,
    )
    usage = models.CharField(
        max_length=120,
        choices=USAGE_CHOICES,
        null=True,
        blank=True,
    )
    .....many more fields....
   

Answer 1

您可以使用两种技术解决您的问题。

数据创建

使用 bulk_create 插入大量记录，如果由于查询量大等原因发生 SQL 错误，则在 bulk_create 中提供 batch_size。

records = []
for elem, ts in request.data['time_series'] :
    records.append(
         TimeSeries(data_json=ts)
    )

# setting batch size t 1000

TimeSeries.objects.bulk_create(records, batch_size=1000)

bulk_create 有一些注意事项，例如它不会产生信号，而其他注意事项请参见 Doc

数据检索

配置rest框架以使用分页默认配置

REST_FRAMEWORK = {
    'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.LimitOffsetPagination',
    'PAGE_SIZE': 100
}

用于自定义配置

class TimeSeriesResultsSetPagination(PageNumberPagination):
    page_size = 50
    page_size_query_param = 'page_size'
    max_page_size = 10000

class BillingRecordsView(generics.ListAPIView):
   serializer_class = TimeSeriesSerializer
   pagination_class = TimeSeriesResultsSetPagination


   def get_queryset(self):
    """Query time-series

    Query by name of network, type of data, creation method and
    source.
    """

    network = self.kwargs["name_network"]

    if TimeSeries.objects.filter(
        network_element__network__name=network,
    ).exists():
        time_series = TimeSeries.objects.filter(
            network_element__network__name=network,
        )
        return time_series
    else:
        raise NotFound()
   

在https://www.django-rest-framework.org/api-guide/pagination/查看其他分页技术

Answer 2

@micromegas 当您的解决方案在理论上是正确的，但是在循环中多次调用 create() 时，我相信这会导致 ConnectionError 异常。

尝试重构为：

big_data_holder = []
for elem, ts in request.data['time_series'] :
    big_data_holder.append(
         TimeSeries(data_json=ts)
    )

# examine the structure 
print(big_data_holder) 

TimeSeries.objects.bulk_create(big_data_holder)

请检查此方法的一些缺点 Django Docs bulk_create