DEV DOG BLOG

สร้าง Decorator Cache ให้กับ Lambda ที่ใช้ Python Runtime ด้วยการทำ Variable Cache (MemoryCache)

ด้วยที่ว่า Lambda ถ้าไม่มีการเรียกใช้งาน ชักระยะนึ่ง Lambda ก็จะถึงทำลายทิ้งไป ทำให้เราสามารถใช้ Variable Cache ช่วยในการเพิ่มความเร็วได้ โดยมีผลกระทบน้อยต่อ ram ที่ใช้งาน ด้วย Code ต่อไปนี้

ไฟล์ MemoryCacheModel.py

from datetime import datetime, timedelta
import os
import uuid
import hashlib

class cached(object):
    def __init__(self, *args, **kwargs):
        self.cached_function_responses = {}
        self.default_max_age = timedelta(seconds=kwargs.get("default_cache_max_age", 0))

    def x_default_dumps(self, x):
        return "LambdaContext" if "LambdaContext" in str(type(x)) else str(x)
        
    def __call__(self, func):
        def inner(*args, **kwargs):
            resourcepath = args[0].get("requestContext",{}).get("path","/")+"/"+args[0].get("requestContext",{}).get("authorizer",{}).get("sub","000")+"/"+args[0].get("queryStringParameters",{}).__str__()

            key = str(uuid.uuid5(uuid.NAMESPACE_DNS,hashlib.md5(str(resourcepath).encode()).hexdigest()))

            max_age = kwargs.get('max_age', self.default_max_age)
            if not max_age or key not in self.cached_function_responses or (datetime.now() - self.cached_function_responses[key]['fetch_time'] > max_age):
                if 'max_age' in kwargs:
                    del kwargs['max_age']
                res = func(*args, **kwargs)
                self.cached_function_responses[key] = {'data': res, 'fetch_time': datetime.now()}
            return self.cached_function_responses[key]['data']
        return inner

วิธีนี่จะเหมาะสำหรับ Function ที่มีการเรียกใช้ในแต่ละครั้ง ใช้เวลานาน และมีจำนวนครั้งในการเรียกใช้งานมากๆ ในระสั่นๆ (ไม่เกิน 15 นาที) และต้องการส่งข้อมูลเดิมออกไปทันที

วิธีนี้จะทำให้ลดระยะเวลาการประมวลผลซ้ำๆ ลงได้ (ให้ใช้เป็นทางเลือกจะดีกว่านะครับ)

วิธีนำไปใช้งาน

ไฟล์ app.py

from MemoryCacheModel import cached

@cached(default_cache_max_age=60*5)
def lambda_handler(event, context):
    
    # long time logic

    return {...}

ตัวอย่างจากที่ ผู้เขียนได้นำไปใช้งาน กับ Logic ต่อไปนี้ โดยลูกค้าไม่ต้องการให้ทำเป็น In-Background Process (ไม่ต้องการให้หลังบ้านทำงานเสร็จแล้วค่อยส่งเมล หรือ บันทึกลง DB)

บน Lambda ที่มีขนาด ram เป็นค่าเริ่มต้นคือ 128mb

1. ประมวลผลไฟล์ Report เพื่อสรุปข้อมูล ประมาณ 10,000 ชุด ของหน่วยงาน 90+ หน่วยงาน
2. สร้างสรุปเป็นไฟล์ excel ด้วย pandas จากข้อมูล 10,000 ชุดแยกเป็นของแต่ละหน่วยงาน
3. อัพโหลดไฟล์ทั้งหมด ไปยัง S3
4. ส่ง URL S3 และ JSON Summary Data (มีบางส่วนที่ไม่แคชด้วยนะครับ) ออกไปผ่านทาง API เพื่อแสดงผล

เรียกใช้งานครั้งแรก ประมาณ 11-12 วินาที

เรียกใช้งานครั้งถัดไประยะเวลาจะลดเหลือเพียง 3-4 วินาที

อีก 1 ตัวอย่าง เป็น การ Get List User โดยมี Logic ต่อไปนี้

บน Lambda ที่มีขนาด ram เป็นค่าเริ่มต้นคือ 128mb

1. ดึงผู้ใช้งานออกมา จาก AWS Cognito ตามจำนวนที่ผู้ใช้งานต้องการ (ในที่นี้เป็น 50 User)
2. นำรายการ User ที่ได้ไปหาข้อมูลใน Database ว่า User คนไหนอยู่ใน Department อะไรบ้างนะเวลาที่ List User ออกมา ล่าสุด 3 ลำดับ
3. เช็คว่า User กำลังใช้งาน Module อะไรอยู่โดยต้องแสดงชื่อ Module ล่าสุด 3 ลำดับ
4. เช็คว่า User ที่ดึงออกมานั้น อยู่ในระบบนานแค่ไหนแล้ว โดยแสดงออกมาเป็นตัวเลข ชัวโมง นาที วินาที
5. เช็คว่าออนไลนหรือออฟไลน์อยู่ โดยใช้การ เทียบระยะเวลา 5 นาทีย้อนหลัง

เรียกใช้งานครั้งแรก 4-5 วินาที

เรียกใช้งานครั้งถัดไประยะเวลาจะลดเหลือเพียง 800m-1200ms

หากใช้งานกับ Function ที่ไม่ได้เป็น Long Time ระยะเวลาอาจจะลดลงได้มากกว่านี้อีกนะครับ ลองเาไปปรับใช้งานกันดูครับ