MongoDB (Document Store)
Python (pymongo)
from pymongo import MongoClient
client = MongoClient('mongodb://localhost:27017')
db = client['ecommerce']
products = db['products']
# 문서 삽입 (스키마 없이 자유롭게)
products.insert_one({
'name': 'MacBook Pro',
'category': 'electronics',
'price': 2499.00,
'specs': {
'cpu': 'M3 Pro',
'ram': '18GB',
'storage': '512GB SSD'
},
'tags': ['laptop', 'apple', 'premium']
})
# 중첩 필드 쿼리
apple_laptops = products.find({
'category': 'electronics',
'specs.cpu': {'$regex': '^M[0-9]'}
})
# Aggregation Pipeline
pipeline = [
{'$match': {'category': 'electronics'}},
{'$group': {
'_id': '$specs.cpu',
'avgPrice': {'$avg': '$price'},
'count': {'$sum': 1}
}},
{'$sort': {'avgPrice': -1}}
]
results = products.aggregate(pipeline)
Redis (Key-Value Store)
Python
import redis
import json
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
# String
r.set('user:1001:session', 'abc123', ex=3600)
# Hash (객체 저장)
r.hset('user:1001', mapping={
'name': 'Alice',
'email': 'alice@example.com',
'plan': 'premium'
})
# Sorted Set (리더보드)
r.zadd('game:leaderboard', {'alice': 1500, 'bob': 1200})
top_players = r.zrevrange('game:leaderboard', 0, 9, withscores=True)
# List (큐)
r.lpush('task:queue', json.dumps({'job': 'send_email', 'to': 'user@example.com'}))
task = r.rpop('task:queue')
Cassandra (Wide-Column Store)
Python (cassandra-driver)
from cassandra.cluster import Cluster
cluster = Cluster(['127.0.0.1'])
session = cluster.connect('analytics')
# 파티션 키 설계가 중요
session.execute("""
CREATE TABLE IF NOT EXISTS user_events (
user_id UUID,
event_date DATE,
event_time TIMESTAMP,
event_type TEXT,
event_data TEXT,
PRIMARY KEY ((user_id, event_date), event_time)
) WITH CLUSTERING ORDER BY (event_time DESC)
""")
# 쓰기 (파티션 단위로 분산)
session.execute("""
INSERT INTO user_events (user_id, event_date, event_time, event_type, event_data)
VALUES (%s, %s, %s, %s, %s)
""", (user_id, today, now, 'page_view', json.dumps(data)))
# 읽기 (파티션 키 필수)
rows = session.execute("""
SELECT * FROM user_events
WHERE user_id = %s AND event_date = %s
ORDER BY event_time DESC
LIMIT 100
""", (user_id, today))
Neo4j (Graph Database)
Cypher
// 노드 생성
CREATE (alice:User {name: 'Alice', age: 28})
CREATE (bob:User {name: 'Bob', age: 32})
CREATE (post:Post {title: 'NoSQL Guide', views: 1500})
// 관계 생성
MATCH (a:User {name: 'Alice'}), (b:User {name: 'Bob'})
CREATE (a)-[:FOLLOWS]->(b)
MATCH (a:User {name: 'Alice'}), (p:Post {title: 'NoSQL Guide'})
CREATE (a)-[:WROTE]->(p)
// 2단계 친구 찾기 (SQL로는 복잡한 쿼리)
MATCH (user:User {name: 'Alice'})-[:FOLLOWS*2]->(fof:User)
WHERE NOT (user)-[:FOLLOWS]->(fof) AND user <> fof
RETURN DISTINCT fof.name AS recommendedFriend
// 최단 경로 탐색
MATCH path = shortestPath(
(a:User {name: 'Alice'})-[:FOLLOWS*]-(b:User {name: 'Charlie'})
)
RETURN path
DynamoDB (Key-Value / Document)
Python (boto3)
import boto3
from boto3.dynamodb.conditions import Key
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Orders')
# 아이템 저장
table.put_item(Item={
'PK': 'USER#1001', # Partition Key
'SK': 'ORDER#2024-01-15', # Sort Key
'total': 299.99,
'items': [
{'product': 'Laptop Stand', 'qty': 1},
{'product': 'USB Cable', 'qty': 2}
],
'status': 'shipped'
})
# 쿼리 (파티션 키 + 정렬 키 범위)
response = table.query(
KeyConditionExpression=
Key('PK').eq('USER#1001') &
Key('SK').begins_with('ORDER#2024')
)
# Single Table Design - GSI로 다양한 접근 패턴
# GSI1: status로 조회, GSI2: 날짜로 조회