DRF Serializer에서 트리 구조 데이터 캐싱 전략

 

Tree 구조와 같이 계층형 데이터 모델 조회 시 데이터 조회 성능을 개선하기 위해 여러 방안을 시도한 경험을 공유합니다.

 

model

class Tree(MPTTModel, TimeStampModel):
    name = CharField(max_length=200)
    order = IntegerField(null=True, blank=True)
    parent = TreeForeignKey(
        "self", on_delete=PROTECT, null=True, 
        blank=True, related_name="children"
    )

 

 

serializer

class TreeSerializer(serializers.ModelSerializer):
	children = serializers.SerializerMethodField(read_only=True)
    
    class Meta:
        model = Tree
        fields = ["id", "children", "name", "order"]
   
    def get_children(self, obj):
    	return TreeSerializer(obj.get_children().order_by('order'), many=True).data

 

 

먼저 저는 django mptt 라이브러리를 활용하여 Tree 구조 모델을 만들었습니다.

위 예시에서 mptt 모델의 내장 메서드인 get_children을 호출해 쉽게 자식 노드를 모두 조회할 수 있지만 계층이 깊어질수록 쿼리 수가 증가하고 직렬화 속도가 느려졌습니다.

 

성능 저하 원인

• 복잡한 데이터 로딩:자식 노드를 재귀적으로 직렬화하면서 N+1 문제가 발생.

• 느린 응답 시간: 트리 구조의 깊이가 깊어질수록 데이터 처리 시간 증가.

• 높은 서버 부하: 동일한 데이터를 반복적으로 직렬화하면서 서버 리소스가 낭비.

 

쿼리 최적화 및 dict 자료구조 활용

쿼리 수를 줄이고 성능을 개선하기 위해 ORM과 dict 자료구조를 활용했습니다.

class Tree(MPTTModel, TimeStampModel):
    name = CharField(max_length=200)
    order = IntegerField(null=True, blank=True)
    parent = TreeForeignKey(
        "self", on_delete=PROTECT, null=True, 
        blank=True, related_name="children"
    )
	
    @staticmethod
    def prefetch_all_children(parent):
        from collections import defaultdict

        prefetched_children = defaultdict(list)
        descendants = parent.get_descendants()
        for descendant in descendants.iterator(chunk_size=1000):
            prefetched_children[descendant.parent_id].append(descendant)
        return prefetched_children


class TreeSerializer(serializers.ModelSerializer):
	children = serializers.SerializerMethodField(read_only=True)
    
    class Meta:
        model = Tree
        fields = ["id", "children", "name", "order"]
   
    def get_children(self, obj):
        if obj.is_leaf:
            return []
        prefetched_children = self.context.get("prefetched_children", None) or Tree.prefetch_all_children(obj)
        children = prefetched_children[obj.id])
        children = sorted(children, key=lambda x: (x.order is None, x.order))
        return TreeSerializer(
            children,
            many=True,
            context={
                "prefetched_children": prefetched_children,
            },
        ).data

 

위 코드를 살펴보면, 먼저 Tree 모델의 prefetch_all_children 메서드에서 mptt 모델의 내장메서드인 get_descendants를 호출해 해당 노드의 모든 하위 노드들을 fetch합니다.

그리고 dict에서 fetch된 노드들은 자신의 부모 노드의 id를 key로 두고 리스트 형태로 value에 담기게 됩니다.

이런식으로 트리 형태를 dict로 치환하였고, serializer에서는 미리 만들어둔 해당 dict를 통해 자식 노드 객체를 가져와 직렬화하도록 개선하였습니다.

 

• prefetch_all_children 메서드로 자식 노드 전체를 한 번의 쿼리로 가져옴.

• 가져온 데이터를 dict 형태로 캐싱하여 부모 ID를 키로 사용해 빠르게 접근.

• Serializer에서 미리 만들어둔 dict를 사용하여 하위 노드를 직렬화.

 

성능 측정

• 기존: 12.879초

• 개선 후: 7.861초

• 약 39% 성능 개선을 달성

성능은 개선되었지만 여전히 추가적인 개선이 필요해보였습니다.

예제에서는 나타나지 않지만, serializer에서 일부 필드로 연관 모델을 조인하는 부분은 여전히 성능 저하에 이슈로 보였습니다.

 

캐싱 활용

여러 방안은 고민하여 redis에 직렬화된 데이터를 값으로 넣어두었다가, 조회 API 호출 시 해당 값을 꺼내어 응답하는 방안을 적용해보기로 했습니다. 해당 tree는 급증하는 데이터가 아니었기 때문에 redis 사용에 큰 무리가 없을 것이라 판단했습니다.

class TreeSerializer(serializers.ModelSerializer):
	children = serializers.SerializerMethodField(read_only=True)
    
    class Meta:
        model = Tree
        fields = ["id", "children", "name", "order"]
   
    def get_children(self, obj):
        if obj.is_leaf:
            return []
        prefetched_children = self.context.get("prefetched_children", None) or Tree.prefetch_all_children(obj)
        children = prefetched_children[obj.id])
        children = sorted(children, key=lambda x: (x.order is None, x.order))
        return TreeSerializer(
            children,
            many=True,
            context={
                "prefetched_children": prefetched_children,
            },
        ).data
       
     def to_representation(self, instance):
        cache_key = "cached_tree_data"
        cached_data = cache.get(cache_key)
        if cached_data:
            return cached_data
        else:
            data = super().to_representation(instance)
            cache.set(cache_key, data, timeout=60 * 60 * 12)
        return data

 

serializer의 to_representation 메서드는 직렬화 데이터를 반환할 때 호출되는 메서드로 이부분에 캐싱 로직을 적용하여 캐싱된 데이터가 반환되도록 하였습니다.

데이터가 주기적으로 최신화될 수 있도록 캐시 만료 시간은 12시간으로 적용하였습니다.

 

• to_representation 메서드에서 캐싱된 데이터가 존재하면 Redis에서 데이터를 가져옴

• 캐싱된 데이터가 없으면 직렬화 후 Redis에 데이터를 저장

 

데이터 최신화 문제 해결

하지만 redis 캐싱 방안은 캐시 만료 전까지 최신화된 데이터를 볼 수 없다는 문제점이 존재했습니다.

저는 데이터 최신화 문제를 해결하기 위해 Django Signals를 활용했습니다.

from django.core.cache import cache
from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver

from tree.models import Tree


@receiver([post_save, post_delete], sender=Tree)
def invalidate_tree_cache(sender, instance, **kwargs):
    cache_key = "cached_tree_data"
    cache.delete(cache_key)

 

django signal을 활용해 모델이 수정되거나 삭제될 경우(post_save, post_delete) 캐싱된 데이터를 삭제하도록 처리했습니다.

 

• post_save: 데이터가 저장되면 캐시를 무효화

• post_delete: 데이터가 삭제되면 캐시를 무효화

 

최종 성능 측정

구분	캐싱 전	쿼리 최적화 및 dict 적용	redis 캐싱 활용
시간(sec)	12.879	7.861	0.0252

• 최종적으로 약 99% 성능 개선을 달성

 

redis를 통해 tree 구조 데이터 조회 성능극적으로 향상시킬 수 있었습니다.

하지만 데이터 수가 대규모이거나 증감 속도가 빠를 경우 redis 메모리 관리에 주의해야 합니다.

 

감사합니다.