django - 减少 Django 序列化时间

Question

我正在查询大约 100,000 行，每行大约 40 列。列是浮点数、整数、日期时间和字符的组合。

查询时间大约为两秒，序列化需要四十秒或更长时间，而响应构建也大约需要两秒。

我想知道如何减少 Django 模型的序列化时间？

这是我的模型:

class TelematicsData(models.Model):
    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)

    device = models.ForeignKey(Device, on_delete=models.CASCADE, null=True)

    created_date = models.DateTimeField(auto_now=True)

    analog_input_01 = models.FloatField(null=True)
    analog_input_02 = models.FloatField(null=True)
    analog_input_03 = models.FloatField(null=True)
    analog_input_04 = models.FloatField(null=True)
    analog_input_05 = models.FloatField(null=True)
    analog_input_06 = models.FloatField(null=True)

    device_temperature = models.FloatField(null=True)
    device_voltage = models.FloatField(null=True)
    vehicle_voltage = models.FloatField(null=True)

    absolute_acceleration = models.FloatField(null=True)
    brake_acceleration = models.FloatField(null=True)
    bump_acceleration = models.FloatField(null=True)
    turn_acceleration = models.FloatField(null=True)
    x_acceleration = models.FloatField(null=True)
    y_acceleration = models.FloatField(null=True)
    z_acceleration = models.FloatField(null=True)

    cell_location_error_meters = models.FloatField(null=True)
    engine_ignition_status = models.NullBooleanField()

    gnss_antenna_status = models.NullBooleanField()
    gnss_type = models.CharField(max_length=20, default='NA')
    gsm_signal_level = models.FloatField(null=True)
    gsm_sim_status = models.NullBooleanField()

    imei = models.CharField(max_length=20, default='NA')
    movement_status = models.NullBooleanField()
    peer = models.CharField(max_length=20, default='NA')

    position_altitude = models.IntegerField(null=True)
    position_direction = models.FloatField(null=True)
    position_hdop = models.IntegerField(null=True)
    position_latitude = models.DecimalField(max_digits=9, decimal_places=6, null=True)
    position_longitude = models.DecimalField(max_digits=9, decimal_places=6, null=True)
    position_point = models.PointField(null=True)
    position_satellites = models.IntegerField(null=True)
    position_speed = models.FloatField(null=True)
    position_valid = models.NullBooleanField()

    shock_event = models.NullBooleanField()
    hardware_version = models.FloatField(null=True)
    software_version = models.FloatField(null=True)

    record_sequence_number = models.IntegerField(null=True)
    timestamp_server = models.IntegerField(null=True)
    timestamp_unix = models.IntegerField(null=True)
    timestamp = models.DateTimeField(null=True)
    vehicle_mileage = models.FloatField(null=True)

    user_data_value_01 = models.FloatField(null=True)
    user_data_value_02 = models.FloatField(null=True)
    user_data_value_03 = models.FloatField(null=True)
    user_data_value_04 = models.FloatField(null=True)
    user_data_value_05 = models.FloatField(null=True)
    user_data_value_06 = models.FloatField(null=True)  
    user_data_value_07 = models.FloatField(null=True)  
    user_data_value_08 = models.FloatField(null=True)

这是序列化程序:

class TelematicsDataSerializer(serializers.ModelSerializer):

    class Meta:
        model = TelematicsData
        geo_field = ('position_point')
        #fields = '__all__'
        exclude = ['id']

Answer 1

简而言之:使用缓存

恕我直言，我认为序列化程序本身没有问题。但问题是数据的大小。

我已经用给定的模型(没有 POSTGIS 部分)对具有 100K 行的序列化程序进行了一些测试，发现在本地机器中平均在 18 秒内生成序列化数据。我测试了 django 的默认值 serializer ，但获得 100K 行大约需要 20 秒。

之间的并排比较DRF 串行器 和 Django 序列化程序 :


所以，作为 FK 关系不是很重要，我也用 prefetch_related 测试过，也没有太大的改善。

所以，我想说，我们需要在其他地方改进。恕我直言，我认为这里的瓶颈是数据库。所以你可以在那里做一些改进，比如 Index Caching (仅供引用:我不是这方面的专家，我不知道它是否可能；也许值得一试)。

但是，还有一个更好的方法是使用 in memory storage来缓存数据。您可以使用 Redis .

在 中存储/检索 100K 行Redis 也比 DB 查询花费的时间少得多(大约 2 秒)。
我本地机器的截图:



你可以这样试试:

首先，将 Json 数据存储在 Redis 中并超时。所以一段时间后，redis 数据会被删除并再次从 DB 加载。

调用一个API时，首先检查Redis中是否存在，如果存在，则从Redis服务

否则，从 Seralizer 提供服务并将 JSON 再次存储在 Redis 中。

编码示例:

import json
import redis


class SomeView(APIView):
    def get(self, request, *args, **kwargs):
        host = getattr(settings, "REDIS_HOST", 'localhost')  # assuming you have configuration in settings.py
        port = getattr(settings, "REDIS_PORT", 6379)
        KEY = getattr(settings, "REDIS_KEY", "TELE_DATA")
        TIME_OUT = getattr(settings, "REDIS_TIMEOUT", 3600)
        pool=redis.StrictRedis(host, port)
        data = pool.get(KEY)
        if not data:
             data = TelematicsDataSerializer(TelematicsData.objects.all(), many=True).data
             pool.set(KEY, json.dumps(data), e=TIME_OUT)
             return Response(data)
        else:
            return Response(json.loads(data))

此解决方案有两个主要缺点。

在超时之间插入的行(假设是一小时)，则不会在响应中发送

如果 Redis 为空，则将需要 40 秒以上的时间将响应发送给用户。

为了克服这些问题，我们可以引入类似 celery 的内容。 ，它将定期更新Redis中的数据。意思是，我们将定义一个新的 celery 任务，它将定期在 Redis 中加载数据并删除旧数据。我们可以这样尝试:

from celery.task.schedules import crontab
from celery.decorators import periodic_task


@periodic_task(run_every=(crontab(minute='*/5')), name="load_cache", ignore_result=True)  # Runs every 5 minute
def load_cache():
    ...
    pool=redis.StrictRedis(host, port)
    json_data = TelematicsDataSerializer(TelematicsData.objects.all(), many=True).data
    pool.set(KEY, json.dumps(data))  # No need for timeout

在 View 中:

class SomeView(APIView):
    def get(self, request, *args, **kwargs):
        host = getattr(settings, "REDIS_HOST", 'localhost')  # assuming you have configuration in settings.py
        port = getattr(settings, "REDIS_PORT", 6379)
        KEY = getattr(settings, "REDIS_KEY", "TELE_DATA")
        TIME_OUT = getattr(settings, "REDIS_TIMEOUT", 3600)
        pool=redis.StrictRedis(host, port)
        data = pool.get(KEY)
        return Response(json.loads(data))

因此，用户将始终从缓存中接收数据。此解决方案还有一个缺点，用户将获得的数据可能没有最新的行(如果它们在 celery 任务的间隔时间之间)。但是假设您想强制 celery 使用 load_cache.apply_async() 重新加载缓存(异步运行)或 load_cache.apply() (同步运行)。

此外，您可以使用 Redis 的许多替代方案进行缓存，例如 memcache , elastic search等等。

实验:

由于数据量很大，也许你可以在存储时压缩数据并在加载时解压数据。但它会降低性能，不确定幅度是多少。你可以这样试试:

压缩

import pickle
import gzip

....

binary_data = pickle.dumps(data)
compressed_data = gzip.compress(binary_data)
pool.set(KEY, compressed_data)  # No need to use JSON Dumps

减压

import pickle
import gzip

....

compressed_data = pool.get(KEY)
binary_data = gzip.decompress(compressed_data)
data = pickle.loads(binary_data)