DynamoDB Time to Live (TTL)#

Keywords: AWS, Amazon, DynamoDB

DynamoDB 的 Time to Live (TTL) 功能可以自动删除过期的项目. 它的价值在当 Item 过期后, DynamoDB 会自动删除 Item, 而不需要应用程序来删除, 也不消耗 WCU. 这个功能是不花钱的.

  • 你需要指定一个 Attribute 作为 TimeToLive 的 expiration time, 凡是当前时间大于这个值的 item 都视为过期数据.

  • 过期的数据不会被立刻删除, 而是会在几天内 (一般是 3 天内) 被 AWS 自动删除.

  • 在还没有被删除之前你依然可以 get, put, update, delete.

  • 你在 Scan 和 Query 的时候默认还是会扫到这些过期数据, 你需要在 query 中指定 filter condition 来过滤掉这些数据. AWS 不会自动帮你过滤这些数据.

下面我们给出了一个示例:

test_dynamodb_ttl.py
 1# -*- coding: utf-8 -*-
 2
 3import time
 4from datetime import datetime, timezone, timedelta
 5
 6import pynamodb_mate.api as pm
 7from boto_session_manager import BotoSesManager
 8
 9
10class Measurement(pm.Model):
11    """
12    我们假设我们的应用场景是用传感器不断地采集测量数据.
13
14    - series: 是一个传感器的唯一标识.
15    - time: 是测量数据的时间戳.
16    - expire_at: 是测量数据的过期时间, 比如我们可以设定测量之后只保留 30 天的数据.
17    """
18
19    class Meta:
20        table_name = "dynamodb-ttl-poc-measurement"
21        region = "us-east-1"
22        billing_mode = pm.constants.PAY_PER_REQUEST_BILLING_MODE
23
24    series = pm.UnicodeAttribute(hash_key=True)
25    time = pm.UTCDateTimeAttribute(range_key=True)
26    # 只要定义了一个 TTL attribute (实际上是一个 UTCDateTimeAttribute),在创建 Table 的时候就会自动启用 TTL 功能
27    expire_at = pm.TTLAttribute()
28
29
30bsm = BotoSesManager(profile_name="bmt_app_dev_us_east_1")
31with bsm.awscli():
32    Measurement._connection = None
33    # 在第一次创建 table 的时候, 会自动打开 ttl 功能
34    if Measurement.exists() is False:
35        Measurement.create_table(wait=True)
36
37
38class EC2Usage(Measurement):
39    """
40    这个 DynamoDB table 只是扩展了 :class:`Measurement` 类, 添加了一些额外的属性.
41    """
42
43    cpu_usage = pm.NumberAttribute()
44    memory_usage = pm.NumberAttribute()
45
46
47def get_utc_now() -> datetime:
48    return datetime.utcnow().replace(tzinfo=timezone.utc)
49
50
51# 清空已有的数据
52EC2Usage.delete_all()
53
54# 插入一条数据
55ec2_inst_id = "i-1a2b3c"
56
57measurement_time = get_utc_now()
58ec2_usage = EC2Usage(
59    series=ec2_inst_id,
60    time=measurement_time,
61    expire_at=measurement_time + timedelta(seconds=5),
62    cpu_usage=0.1,
63    memory_usage=0.2,
64)
65ec2_usage.save()
66
67# 每秒查询一次数据, 从第 6 次开始, 就不会再查询到数据了
68for i in range(1, 1 + 10):
69    time.sleep(1)
70    now = get_utc_now()
71    print(f"at {i}th second")
72    ec2_usage_list = EC2Usage.iter_query(
73        ec2_inst_id,
74        filter_condition=EC2Usage.expire_at >= now,  # 用 filter 来过滤掉已经过期的数据
75    ).all()
76    print(f"  {ec2_usage_list = }")

Reference#

https://pynamodb.readthedocs.io/en/stable/api.html#pynamodb.attributes.TTLAttribute