목차
- boto3 응답 구조부터 이해하기
- filter()와 sorted()에 lambda 붙이는 기본 문법
- EC2 인스턴스 필터링·정렬 실전 패턴
- S3 객체 필터링·정렬 패턴
- DynamoDB scan/query 결과 후처리
- 서비스별 boto3 응답 데이터 필터링 정렬 비교
- 실무 체크리스트
- Lambda 함수(AWS)에서 쓸 때 주의할 점
- 서버사이드 vs 클라이언트사이드, 어디서 필터링할 것인가
- 다음 단계
boto3로 AWS 리소스를 조회하면 응답은 대부분 dict 리스트 형태다. boto3 응답 데이터 필터링 정렬은 결국 이 리스트에서 원하는 항목만 골라내고, 특정 키 기준으로 줄 세우는 작업이다. describe_instances(), list_objects_v2(), scan() — 어떤 API든 리턴 구조가 비슷하기 때문에 패턴 하나를 익히면 거의 모든 서비스에 적용된다.
문제는 boto3 자체에 클라이언트 사이드 필터링·정렬 메서드가 없다는 점이다. 서버사이드 필터(Filters 파라미터)가 있긴 하지만, 모든 조건을 서버에서 걸 수 있는 건 아니고 서비스마다 지원 범위가 다르다. 결국 Python 내장 함수인 filter(), sorted()와 lambda를 조합하는 게 가장 범용적인 방법이 된다.
boto3 응답 구조부터 이해하기
boto3 API 응답은 JSON을 Python dict로 변환한 형태다. EC2의 describe_instances()를 예로 들면 이런 구조가 온다:
{
'Reservations': [
{
'Instances': [
{
'InstanceId': 'i-0abc123',
'State': {'Name': 'running'},
'LaunchTime': datetime(2026, 3, 15, 9, 0, 0),
'Tags': [{'Key': 'Name', 'Value': 'web-prod-01'}]
}
]
}
]
}S3의 list_objects_v2()는 다르다:
{
'Contents': [
{
'Key': 'logs/2026-04-01.gz',
'Size': 1048576,
'LastModified': datetime(2026, 4, 1, 0, 0, 0)
}
]
}DynamoDB scan()도 또 다르다:
{
'Items': [
{
'user_id': {'S': 'u-001'},
'score': {'N': '85'},
'created_at': {'S': '2026-04-10T12:00:00Z'}
}
]
}세 서비스 모두 최상위 키 아래에 dict 리스트가 들어있다는 공통점이 있다. Reservations[*].Instances[*], Contents, Items — 키 이름만 다를 뿐 구조는 같다. 이 리스트를 꺼내서 filter()와 sorted()에 넘기면 된다.
거의 모든 boto3 List/Describe/Scan API는 `{‘최상위키’: [dict, dict, …]}` 형태로 응답한다. 최상위 키 이름은 서비스마다 다르지만, 내부 처리 로직은 동일하게 적용할 수 있다.
Python 내장 함수 두 개면 충분하다.
filter()로 조건에 맞는 항목만 남기기
filter(function, iterable) — function이 True를 반환하는 항목만 남긴다.
instances = [
{'InstanceId': 'i-001', 'State': {'Name': 'running'}},
{'InstanceId': 'i-002', 'State': {'Name': 'stopped'}},
{'InstanceId': 'i-003', 'State': {'Name': 'running'}},
]
running = list(filter(lambda x: x['State']['Name'] == 'running', instances))filter()는 iterator를 반환하므로 list()로 감싸야 리스트가 된다. 리스트 컴프리헨션으로 같은 결과를 얻을 수도 있다:
running = [x for x in instances if x['State']['Name'] == 'running']둘 중 뭘 쓸지는 취향이다. 단일 조건이면 컴프리헨션이 읽기 쉽고, 조건이 복잡하거나 재사용할 함수가 있으면 filter()가 낫다.
sorted()로 특정 키 기준 정렬
sorted(iterable, key=function) — key 함수가 반환하는 값 기준으로 정렬한다.
objects = [
{'Key': 'a.log', 'Size': 500},
{'Key': 'b.log', 'Size': 100},
{'Key': 'c.log', 'Size': 300},
]
by_size_desc = sorted(objects, key=lambda x: x['Size'], reverse=True)reverse=True를 붙이면 내림차순. 기본값은 오름차순이다.
체이닝: filter 후 sorted
두 함수를 연결하면 "조건 필터링 → 정렬"이 한 줄에 끝난다:
result = sorted(
filter(lambda x: x['Size'] > 200, objects),
key=lambda x: x['Size'],
reverse=True
)이 패턴이 boto3 응답 데이터 필터링 정렬의 핵심 골격이다. 아래 서비스별 예제는 전부 이 구조의 변형이다.
EC2 인스턴스 필터링·정렬 실전 패턴
EC2는 describe_instances() 응답이 Reservations → Instances 이중 중첩이라 먼저 평탄화(flatten)가 필요하다.
인스턴스 리스트 평탄화
import boto3
ec2 = boto3.client('ec2', region_name='ap-northeast-2')
response = ec2.describe_instances()
instances = [
inst
for res in response['Reservations']
for inst in res['Instances']
]이렇게 하면 instances는 단순한 dict 리스트가 된다. 이후 filter()와 sorted()를 바로 적용할 수 있다.
상태별 필터링
running 상태 인스턴스만 가져오기:
running = list(filter(
lambda x: x['State']['Name'] == 'running',
instances
))서버사이드 필터로도 가능하다. describe_instances(Filters=[{'Name': 'instance-state-name', 'Values': ['running']}]) — 이 경우 API 호출 자체에서 결과를 줄이므로 네트워크 비용이 줄어든다. 인스턴스 수가 수백 개 이상이면 서버사이드 필터를 우선 쓰는 게 맞다.
클라이언트 사이드 필터가 필요한 경우는 서버에서 지원하지 않는 조건을 걸 때다. 예를 들어 "태그 값에 특정 패턴이 포함된 인스턴스"는 서버 필터로 정확히 매칭할 수 있지만, 정규식 매칭은 안 된다.
태그 기반 필터링
EC2 태그는 [{'Key': 'Name', 'Value': 'web-01'}, ...] 형태라 접근이 번거롭다. 헬퍼 함수를 하나 만들어두면 편하다:
def get_tag(instance, key, default=''):
tags = instance.get('Tags', [])
for tag in tags:
if tag['Key'] == key:
return tag['Value']
return default
prod_instances = list(filter(
lambda x: get_tag(x, 'Environment') == 'production',
instances
))[!tip] 태그 접근 헬퍼는 한번 만들어두면 EC2 외에 RDS, ELB 등 태그 구조가 같은 모든 서비스에서 재사용된다.
LaunchTime 기준 정렬
최근에 시작된 인스턴스 순으로 정렬:
recent_first = sorted(
running,
key=lambda x: x['LaunchTime'],
reverse=True
)
for inst in recent_first:
print(f"{inst['InstanceId']} - {get_tag(inst, 'Name')} - {inst['LaunchTime']}")LaunchTime은 datetime 객체라 비교 연산이 바로 된다. 문자열 변환 없이 sorted()에 넘기면 된다.
다중 조건 정렬도 가능하다. 상태 → LaunchTime 순으로 정렬하려면 key에서 튜플을 반환하면 된다:
multi_sorted = sorted(
instances,
key=lambda x: (x['State']['Name'], x['LaunchTime']),
reverse=True
)Python의 튜플 비교는 첫 번째 요소부터 순서대로 비교하므로, 이 코드는 상태 이름 역순 → 같은 상태 내에서 LaunchTime 역순으로 정렬한다.
S3 객체 필터링·정렬 패턴
S3의 list_objects_v2()는 EC2보다 응답 구조가 단순하다. Contents 키 아래에 객체 리스트가 바로 있다.
s3 = boto3.client('s3')
response = s3.list_objects_v2(Bucket='my-log-bucket', Prefix='logs/')
objects = response.get('Contents', [])response.get('Contents', []) — 버킷이 비어있거나 매칭 객체가 없으면 Contents 키 자체가 없다. .get()으로 빈 리스트를 기본값으로 줘야 KeyError를 피할 수 있다.
크기 기반 필터링
100MB 이상인 파일만 골라내기:
large_files = list(filter(
lambda x: x['Size'] > 100 * 1024 * 1024,
objects
))특정 확장자 필터링:
gz_files = list(filter(
lambda x: x['Key'].endswith('.gz'),
objects
))최종 수정일 기준 정렬과 조합
가장 최근 수정된 .gz 파일 10개:
recent_gz = sorted(
filter(lambda x: x['Key'].endswith('.gz'), objects),
key=lambda x: x['LastModified'],
reverse=True
)[:10]
for obj in recent_gz:
size_mb = obj['Size'] / (1024 * 1024)
print(f"{obj['Key']} - {size_mb:.1f}MB - {obj['LastModified']}")sorted() 결과에 슬라이싱([:10])을 붙여 상위 N개만 취하는 패턴은 자주 쓰인다.
`list_objects_v2()`는 한 번에 최대 1,000개만 반환한다. 객체가 그 이상이면 `NextContinuationToken`으로 반복 호출해야 한다. 전체 객체를 모은 뒤 필터링·정렬해야 결과가 정확하다. 페이지네이터(`s3.get_paginator(‘list_objects_v2’)`)를 쓰면 이 과정을 자동화할 수 있다.
객체가 1,000개를 넘을 때:
paginator = s3.get_paginator('list_objects_v2')
all_objects = []
for page in paginator.paginate(Bucket='my-log-bucket', Prefix='logs/'):
all_objects.extend(page.get('Contents', []))
old_large = sorted(
filter(lambda x: x['Size'] > 50 * 1024 * 1024, all_objects),
key=lambda x: x['LastModified']
)이 코드는 전체 객체를 메모리에 올린다. 수백만 개 객체가 있는 버킷이라면 Prefix를 좁히거나 S3 Inventory를 쓰는 게 낫다.
DynamoDB scan/query 결과 후처리
DynamoDB는 두 가지 면에서 EC2, S3와 다르다. 첫째, 응답 항목의 값이 {'S': 'hello'}, {'N': '42'} 같은 타입 디스크립터 형태다. 둘째, scan()은 테이블 전체를 읽으므로 비용과 시간이 크다.
타입 디스크립터 처리
boto3.resource('dynamodb')를 쓰면 자동으로 Python 타입으로 변환해준다. client 인터페이스를 쓸 때는 직접 변환해야 한다:
# resource 인터페이스 — 타입 자동 변환
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('users')
response = table.scan()
items = response['Items']
# items[0] = {'user_id': 'u-001', 'score': 85, 'status': 'active'}# client 인터페이스 — 타입 디스크립터 그대로
dynamodb = boto3.client('dynamodb')
response = dynamodb.scan(TableName='users')
items = response['Items']
# items[0] = {'user_id': {'S': 'u-001'}, 'score': {'N': '85'}, 'status': {'S': 'active'}}필터링·정렬 관점에서는 resource 인터페이스가 훨씬 편하다. score 값을 비교할 때 item['score']로 바로 쓸 수 있으니까.
score 기준 필터링과 정렬
resource 인터페이스 기준:
table = boto3.resource('dynamodb').Table('users')
response = table.scan()
items = response['Items']
active_high = sorted(
filter(
lambda x: x.get('status') == 'active' and x.get('score', 0) >= 80,
items
),
key=lambda x: x.get('score', 0),
reverse=True
).get() 사용에 주의해야 한다. DynamoDB는 스키마리스이므로 특정 항목에 score 속성이 아예 없을 수 있다. x['score']로 접근하면 KeyError가 나고, x.get('score', 0)으로 기본값을 줘야 안전하다.
scan 페이지네이션
DynamoDB scan()도 한 번에 1MB까지만 반환한다:
all_items = []
response = table.scan()
all_items.extend(response['Items'])
while 'LastEvaluatedKey' in response:
response = table.scan(ExclusiveStartKey=response['LastEvaluatedKey'])
all_items.extend(response['Items'])
filtered = list(filter(lambda x: x.get('status') == 'active', all_items))`scan()`은 테이블 전체를 읽는다. 항목 수만큼 RCU를 소비하므로 프로덕션 테이블에서 무분별하게 실행하면 안 된다. 가능하면 `query()`로 파티션 키 조건을 걸거나, `FilterExpression`으로 서버사이드 필터링을 먼저 적용하는 편이 맞다.
서비스별 boto3 응답 데이터 필터링 정렬 비교
| 항목 | EC2 | S3 | DynamoDB |
|---|---|---|---|
| API | describe_instances() |
list_objects_v2() |
scan() / query() |
| 응답 키 | Reservations[*].Instances[*] |
Contents |
Items |
| 중첩 깊이 | 2단계 (평탄화 필요) | 1단계 | 1단계 |
| 서버 필터 | Filters 파라미터 |
Prefix 만 가능 |
FilterExpression |
| 페이지네이션 | NextToken |
ContinuationToken |
LastEvaluatedKey |
| 정렬 지원 | 서버 정렬 없음 | 키 이름 사전순 고정 | SortKey 정렬 (query만) |
| 값 타입 | Python 네이티브 | Python 네이티브 | resource는 네이티브, client는 타입 디스크립터 |
이 표에서 주목할 점은 서버사이드 정렬을 지원하는 서비스가 거의 없다는 것이다. DynamoDB의 query()만 SortKey 기준 정렬을 지원하고, EC2와 S3는 클라이언트에서 sorted()를 써야 한다. boto3 응답 데이터 필터링 정렬에 lambda 패턴이 필수인 이유가 여기에 있다.
실무 체크리스트
자동화 스크립트에서 boto3 응답을 처리할 때 빠뜨리기 쉬운 항목을 정리한다.
필터링 전 확인사항
- 빈 응답 처리:
response.get('Contents', [])식으로 키 부재에 대비했는가. S3 빈 버킷, EC2 인스턴스 0개 등의 상황에서KeyError가 나지 않아야 한다. - 페이지네이션: 결과가 한 페이지에 다 오는가. EC2는 기본 1,000개, S3는 1,000개, DynamoDB는 1MB 제한이 있다. 전체 데이터를 모으지 않고 첫 페이지만 필터링하면 결과가 불완전해진다.
- 서버 필터 우선: 클라이언트 필터링 전에 서버사이드 필터로 줄일 수 있는 부분이 있는지 확인. 네트워크 전송량과 처리 시간 모두 줄어든다.
- None 값 방어:
lambda x: x.get('Score', 0) >= 80—.get()에 기본값을 주지 않으면None과 숫자 비교에서TypeError가 난다.
정렬 시 확인사항
- datetime 직접 비교: boto3가 반환하는
LaunchTime,LastModified는datetime객체다. 문자열 변환 없이sorted()의 key로 바로 쓸 수 있다. - 다중 키 정렬:
key=lambda x: (x['state'], x['time'])— 튜플 반환으로 우선순위 정렬. 오름/내림 방향이 키마다 다르면 숫자 키에-를 붙여서 역순으로 만든다:key=lambda x: (-x['score'], x['name']). - 안정 정렬: Python의
sorted()는 Timsort 기반 안정 정렬(stable sort)이다. 같은 키 값을 가진 항목은 원래 순서가 유지된다.
필터링 결과가 예상과 다를 때는 lambda 함수를 따로 빼서 중간 값을 출력하면 원인을 빨리 찾는다. `def my_filter(x): val = x.get(‘State’, {}); print(val); return val.get(‘Name’) == ‘running’` 식으로 임시 함수를 만들어 쓰면 된다.
Python lambda 키워드와 AWS Lambda 서비스 이름이 같아서 혼동하기 쉽다. 여기서는 AWS Lambda 함수 안에서 boto3 필터링 코드를 실행할 때의 주의사항을 다룬다.
메모리와 실행 시간 제약이 핵심이다. Lambda 함수는 기본 128MB 메모리, 최대 15분 타임아웃이다. EC2 인스턴스 수십 개를 필터링하는 건 문제없지만, S3에서 수십만 개 객체를 전부 메모리에 올려서 정렬하려면 메모리가 부족할 수 있다.
import boto3
def lambda_handler(event, context):
ec2 = boto3.client('ec2', region_name='ap-northeast-2')
response = ec2.describe_instances(
Filters=[{'Name': 'instance-state-name', 'Values': ['running']}]
)
instances = [
inst
for res in response['Reservations']
for inst in res['Instances']
]
no_name_tag = list(filter(
lambda x: not any(t['Key'] == 'Name' for t in x.get('Tags', [])),
instances
))
return {
'statusCode': 200,
'body': {
'untagged_count': len(no_name_tag),
'instance_ids': [i['InstanceId'] for i in no_name_tag]
}
}이 예제는 Name 태그가 없는 running 인스턴스를 찾는다. 서버 필터(Filters)로 running 상태를 먼저 거르고, 클라이언트 필터(filter())로 태그 조건을 건다. 서버에서 줄일 수 있는 건 서버에서 줄이는 게 원칙이다.
Lambda 핸들러 바깥에서 `boto3.client()`를 생성하면 컨테이너 재사용 시 매번 새 연결을 만들지 않는다. 콜드 스타트 이후 호출에서 응답이 빨라진다.
def filter_pages(paginator, bucket, predicate):
for page in paginator.paginate(Bucket=bucket):
for obj in page.get('Contents', []):
if predicate(obj):
yield obj정렬이 필요 없고 필터링만 하면 되는 경우, 이 방식이 메모리 측면에서 유리하다. 정렬은 전체 데이터가 메모리에 있어야 하므로 제너레이터와 양립하기 어렵다.
서버사이드 vs 클라이언트사이드, 어디서 필터링할 것인가
이 판단 기준은 단순하다.
서버사이드를 먼저 쓴다. 서버에서 지원하는 필터 조건이면 API 파라미터로 넘겨서 결과 자체를 줄이는 게 무조건 낫다. 네트워크 전송량이 줄고, 클라이언트 메모리도 아끼고, 처리 시간도 짧아진다.
클라이언트사이드는 서버가 못 하는 조건에 쓴다. 정규식 매칭, 복수 필드 조합 비교, 계산 기반 조건(Size > 평균 크기) 같은 건 서버 필터로 표현할 수 없다. 이런 조건이 필요할 때만 Python filter()를 쓰면 된다.
실무에서는 둘을 조합하는 경우가 대부분이다:
# 1단계: 서버사이드 — running 인스턴스만 가져오기
response = ec2.describe_instances(
Filters=[{'Name': 'instance-state-name', 'Values': ['running']}]
)
# 2단계: 평탄화
instances = [i for r in response['Reservations'] for i in r['Instances']]
# 3단계: 클라이언트사이드 — 48시간 이상 실행된 것만
from datetime import datetime, timezone, timedelta
cutoff = datetime.now(timezone.utc) - timedelta(hours=48)
long_running = sorted(
filter(lambda x: x['LaunchTime'] < cutoff, instances),
key=lambda x: x['LaunchTime']
)이 코드는 48시간 넘게 돌아가는 인스턴스를 LaunchTime 오름차순으로 정렬한다. 가장 오래된 인스턴스가 맨 위에 온다. 모니터링 스크립트나 정리 자동화에서 자주 보이는 패턴이다.
서버와 클라이언트 필터를 잘 나누는 것이 boto3 응답 데이터 필터링 정렬의 실전 핵심이라고 볼 수 있다. API 호출 한 번에 끝낼 수 있는 건 서버에서 끝내고, 남은 세밀한 조건만 lambda로 처리하는 습관을 들이면 스크립트가 빨라지고 비용도 줄어든다.
다음 단계
boto3 응답 필터링·정렬 패턴이 잡혔으면, 다음으로 볼 만한 주제는 boto3 Paginator 공식 가이드다. 페이지네이션을 제대로 쓰면 대량 데이터 처리가 깔끔해진다. 그리고 JMESPath도 한번 살펴볼 만하다. boto3 응답에서 중첩 구조를 쿼리하는 데 쓸 수 있는데, EC2의 Reservations[*].Instances[*] 같은 이중 중첩을 한 줄로 평탄화할 수 있다. AWS CLI의 --query 파라미터가 내부적으로 JMESPath를 쓰고 있기도 하다.
비용 최적화 관점에서는 AWS Cost Explorer API를 boto3로 호출해서 비용 데이터를 필터링·정렬하는 것도 같은 패턴으로 가능하다. GetCostAndUsage 응답도 결국 dict 리스트이므로, 이 글에서 다룬 filter()와 sorted() 조합이 그대로 적용된다.