加速 Django 数据库函数以对缺失值进行地理插值

Posted 2023-02-24

【中文标题】加速 Django 数据库函数以对缺失值进行地理插值

【英文标题】：Speeding up a Django database function for geographic interpolation of missing values

【发布时间】：2018-09-09 06:37:19

【问题描述】：

我有一个大型的商业物业地址数据库（大约 500 万行），其中 200,000 个缺少楼面面积。这些房产是按行业分类的，我知道每个房产的租金。

我对缺失的建筑面积进行插值的方法是在建筑面积未知的房产的指定半径内过滤出类似分类的房产，然后根据附近房产的成本/平方米的中位数计算建筑面积。

最初，我使用pandas 来解决这个问题，但随着数据集变得越来越大（即使使用group_by），这已经成为问题。它经常超出可用内存并停止。当它工作时，大约需要 3 个小时才能完成。

我正在测试是否可以在数据库中执行相同的任务。我为径向填充编写的函数如下：

def _radial_fill(self):
    # Initial query selecting all latest locations, and excluding null rental valuations
    q = Location.objects.order_by("locode","-update_cycle") \
                        .distinct("locode")
    # Chained Q objects to use in filter
    f = Q(rental_valuation__isnull=False) & \
        Q(use_category__grouped_by__isnull=False) & \
        Q(pc__isnull=False)
    # All property categories at subgroup level
    for c in LocationCategory.objects.filter(use_category="SGP").all():
        # Start looking for appropriate interpolation locations
        fc = f & Q(use_category__grouped_by=c)
        for l in q.filter(fc & Q(floor_area__isnull=True)).all():
            r_degree = 0
            while True:
                # Default Distance is metres, so multiply accordingly
                r = (constants.BOUNDS**r_degree)*1000 # metres
                ql = q.annotate(distance=Distance("pc__point", l.pc.point)) \
                      .filter(fc & Q(floor_area__isnull=False) & Q(distance__lte=r)) \
                      .values("rental_valuation", "floor_area")
                if len(ql) < constants.LOWER_RANGE:
                    if r > constants.UPPER_RADIUS*1000:
                        # Further than the longest possible distance
                        break
                    r_degree += 1
                else:
                    m = median([x["rental_valuation"]/x["floor_area"]
                                for x in ql if x["floor_area"] > 0.0])
                    l.floor_area = l.rental_valuation / m
                    l.save()
                    break

我的问题是这个函数需要 6 天才能运行。必须有更快的方法，对吧？我确定我做错了什么......

型号如下：

class LocationCategory(models.Model):
    # Category types
    GRP = "GRP"
    SGP = "SGP"
    UST = "UST"
    CATEGORIES = (
        (GRP, "Group"),
        (SGP, "Sub-group"),
        (UST, "Use type"),
    )
    slug = models.CharField(max_length=24, primary_key=True, unique=True)
    usecode = models.CharField(max_length=14, db_index=True)
    use_category = models.CharField(max_length=3, choices=CATEGORIES,
                                    db_index=True, default=UST)
    grouped_by = models.ForeignKey("self", null=True, blank=True,
                                   on_delete=models.SET_NULL,
                                   related_name="category_by_group")

class Location(models.Model):
    # Hereditament identity and location
    slug = models.CharField(max_length=24, db_index=True)
    locode = models.CharField(max_length=14, db_index=True)
    pc = models.ForeignKey(Postcode, null=True, blank=True,
                           on_delete=models.SET_NULL,
                           related_name="locations_by_pc")
    use_category = models.ForeignKey(LocationCategory, null=True, blank=True,
                                     on_delete=models.SET_NULL,
                                     related_name="locations_by_category")
    # History fields
    update_cycle = models.CharField(max_length=14, db_index=True)
    # Location-specific econometric data
    floor_area = models.FloatField(blank=True, null=True)
    rental_valuation = models.FloatField(blank=True, null=True)

class Postcode(models.Model):
    pc = models.CharField(max_length=7, primary_key=True, unique=True) # Postcode excl space
    pcs = models.CharField(max_length=8, unique=True)                  # Postcode incl space
    # http://spatialreference.org/ref/epsg/osgb-1936-british-national-grid/
    point = models.PointField(srid=4326)

使用 Django 2.0 和 Postgresql 10

更新

通过以下代码更改，我在运行时实现了 35% 的改进：

# Initial query selecting all latest locations, and excluding null rental valuations
q = Location.objects.order_by("slug","-update_cycle") \
                    .distinct("slug")
# Chained Q objects to use in filter
f = Q(rental_valuation__isnull=False) & \
    Q(pc__isnull=False) & \
    Q(use_category__grouped_by_id=category_id)
# All property categories at subgroup level
# Start looking for appropriate interpolation locations
for l in q.filter(f & Q(floor_area__isnull=True)).all().iterator():
    r = q.filter(f & Q(floor_area__isnull=False) & ~Q(floor_area=0.0))
    rl = Location.objects.filter(id__in = r).annotate(distance=D("pc__point", l.pc.point)) \
                                            .order_by("distance")[:constants.LOWER_RANGE] \
                                            .annotate(floor_ratio = F("rental_valuation")/
                                                                    F("floor_area")) \
                                            .values("floor_ratio")
    if len(rl) == constants.LOWER_RANGE:
        m = median([h["floor_ratio"] for h in rl])
        l.floor_area = l.rental_valuation / m
        l.save()

id__in=r 效率低下，但它似乎是在对新注释添加和排序时维护distinct 查询集的唯一方法。鉴于在 r 查询中可以返回大约 100,000 行，因此在此处应用的任何注释以及随后按距离排序都可能需要很长时间。

但是……我在尝试实现子查询功能时遇到了很多问题。 AttributeError: 'ResolvedOuterRef' object has no attribute '_output_field_or_none' 我认为与注释有关，但我找不到太多。

相关重构代码为：

rl = Location.objects.filter(id__in = r).annotate(distance=D("pc__point", OuterRef('pc__point'))) \
                                        .order_by("distance")[:constants.LOWER_RANGE] \
                                        .annotate(floor_ratio = F("rental_valuation")/
                                                                F("floor_area")) \
                                        .distinct("floor_ratio")

和：

l.update(floor_area= F("rental_valuation") / CustomAVG(Subquery(locs),0))

我可以看到这种方法应该非常有效，但要做到正确似乎有点超出我的技能水平。

【参考方案1】：

您可以使用（大部分）经过优化的 Django 内置查询方法来简化您的方法。更具体地说，我们将使用：

Subquery 和 OuterRef 方法（版本 >= 1.11）。 annotation 和 AVG 来自 Django aggregation。 dwithin 查找。 F() 表达式（F() 的详细用例可以在我的 QA 样式示例中找到：How to execute arithmetic operations between Model fields in django

---

我们将创建一个自定义 Aggregate 类来应用我们的 AVG 函数（该方法的灵感来自这个出色的答案：Django 1.11 Annotating a Subquery Aggregate）

class CustomAVG(Subquery):
    template = "(SELECT AVG(area_value) FROM (%(subquery)s))"
    output_field = models.FloatField()

我们将使用它来计算以下平均值：

for location in Location.objects.filter(rental_valuation__isnull=True):
    location.update(
        rental_valuation=CustomAVG(
            Subquery(
                Location.objects.filter(
                    pc__point__dwithin=(OuterRef('pc__point'), D(m=1000)),
                    rental_valuation__isnull=False
                ).annotate(area_value=F('rental_valuation')/F('floor_area'))
                .distinct('area_value')
            )
        )
    )

上述细分：

我们收集所有没有rental_valuation 的Location 对象，然后“通过”列表。

子查询：

我们从当前位置点选择radius=1000m 圆圈内的Location 对象（根据需要更改），然后annotate on他们计算成本/平方米（使用F() 获取每个对象的rental_valuation 和floor_area 列的值），作为名为@9​​87654347@ 的列。为了获得更准确的结果，我们仅选择此列的不同值。 我们将CustomAVG 应用到Subquery 并更新我们当前的位置rental_valuation。

【讨论】：

谢谢。我还尝试将各个循环拆分为 Celery 进程，看看是否有任何效果。采纳您的建议后，我会尝试发布一些更新时间...

一个问题是，需要通过将计算出的rental_valuation 限制在仅位置计数大于constants.LOWER_RANGE 的位置来确保某种统计完整性。即仅使用 1 或 2 个其他位置来推断“area_value”是有风险的，我想使用尽可能小的区域来计算能够产生该平均值的属性的数量。有什么想法吗？

@Turukawa 对于第一部分，您可以使用条件表达式：docs.djangoproject.com/en/2.0/ref/models/…。对于第二部分，您需要根据您的数据集定义可能的最小范围。

@Turukawa 你觉得这个答案有帮助吗？

说实话，我还在想办法实现它……到目前为止，我已经实现了 35% 的时间改进（6 天到 4 天），但没有子查询。我遇到的问题是在应用distinct（这似乎需要第二次查询）之后对注释进行排序，然后弄清楚如何获得中位数而不是AVG，因为我需要丢弃异常值（尚未解决）.. . 仍在努力，并会在几天后回复您。