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