Docker로 ScyllaDB 실행
# 단일 노드 ScyllaDB 실행
docker run --name scylla -d \
-p 9042:9042 \
-p 9160:9160 \
scylladb/scylla
# 클러스터 노드 상태 확인
docker exec -it scylla nodetool status
# cqlsh 접속
docker exec -it scylla cqlsh
# 3노드 클러스터 구성 (docker-compose.yml)
# version: '3'
# services:
# scylla-node1:
# image: scylladb/scylla
# command: --seeds=scylla-node1
# scylla-node2:
# image: scylladb/scylla
# command: --seeds=scylla-node1
# scylla-node3:
# image: scylladb/scylla
# command: --seeds=scylla-node1
CQL 스키마 및 쿼리
-- Keyspace 생성 (Replication Factor 3)
CREATE KEYSPACE IF NOT EXISTS my_app
WITH replication = {
'class': 'NetworkTopologyStrategy',
'datacenter1': 3
}
AND durable_writes = true;
USE my_app;
-- 사용자 활동 테이블 (시계열 데이터)
CREATE TABLE user_activities (
user_id UUID,
activity_date DATE,
activity_time TIMESTAMP,
activity_type TEXT,
details MAP,
PRIMARY KEY ((user_id, activity_date), activity_time)
) WITH CLUSTERING ORDER BY (activity_time DESC)
AND compaction = {
'class': 'TimeWindowCompactionStrategy',
'compaction_window_unit': 'DAYS',
'compaction_window_size': 1
};
-- 보조 인덱스 (activity_type으로 조회)
CREATE INDEX idx_activity_type ON user_activities (activity_type);
-- Materialized View (타입별 조회용)
CREATE MATERIALIZED VIEW activities_by_type AS
SELECT * FROM user_activities
WHERE activity_type IS NOT NULL
AND user_id IS NOT NULL
AND activity_date IS NOT NULL
AND activity_time IS NOT NULL
PRIMARY KEY ((activity_type, activity_date), activity_time, user_id);
-- 데이터 삽입
INSERT INTO user_activities (user_id, activity_date, activity_time, activity_type, details)
VALUES (
uuid(),
'2024-01-15',
'2024-01-15 10:30:00+0000',
'login',
{'ip': '192.168.1.1', 'device': 'mobile'}
);
-- TTL 적용 (30일 후 자동 삭제)
INSERT INTO user_activities (user_id, activity_date, activity_time, activity_type, details)
VALUES (uuid(), '2024-01-15', toTimestamp(now()), 'pageview', {'page': '/home'})
USING TTL 2592000;
-- 범위 쿼리 (특정 날짜의 활동)
SELECT * FROM user_activities
WHERE user_id = 123e4567-e89b-12d3-a456-426614174000
AND activity_date = '2024-01-15'
AND activity_time >= '2024-01-15 00:00:00+0000'
LIMIT 100;
-- Lightweight Transaction (조건부 삽입)
INSERT INTO user_activities (user_id, activity_date, activity_time, activity_type, details)
VALUES (uuid(), '2024-01-15', toTimestamp(now()), 'signup', {})
IF NOT EXISTS;
Python 드라이버 사용 예제
from cassandra.cluster import Cluster
from cassandra.auth import PlainTextAuthProvider
from cassandra.policies import DCAwareRoundRobinPolicy, TokenAwarePolicy
from cassandra.query import SimpleStatement, ConsistencyLevel
from cassandra import WriteTimeout, ReadTimeout
import uuid
from datetime import datetime, date
# 클러스터 연결 설정
auth_provider = PlainTextAuthProvider(
username='scylla',
password='password'
)
cluster = Cluster(
contact_points=['scylla-node1', 'scylla-node2', 'scylla-node3'],
port=9042,
auth_provider=auth_provider,
load_balancing_policy=TokenAwarePolicy(
DCAwareRoundRobinPolicy(local_dc='datacenter1')
),
protocol_version=4
)
session = cluster.connect('my_app')
# Prepared Statement (성능 최적화)
insert_stmt = session.prepare("""
INSERT INTO user_activities
(user_id, activity_date, activity_time, activity_type, details)
VALUES (?, ?, ?, ?, ?)
""")
select_stmt = session.prepare("""
SELECT * FROM user_activities
WHERE user_id = ? AND activity_date = ?
LIMIT ?
""")
# 데이터 삽입 (배치)
from cassandra.query import BatchStatement, BatchType
def insert_activities_batch(activities: list):
"""여러 활동을 배치로 삽입"""
batch = BatchStatement(batch_type=BatchType.UNLOGGED)
for activity in activities:
batch.add(insert_stmt, (
activity['user_id'],
activity['date'],
activity['timestamp'],
activity['type'],
activity.get('details', {})
))
try:
session.execute(batch)
except WriteTimeout:
# 개별 재시도 로직
for activity in activities:
session.execute(insert_stmt, (
activity['user_id'],
activity['date'],
activity['timestamp'],
activity['type'],
activity.get('details', {})
))
# 비동기 쿼리
from cassandra.concurrent import execute_concurrent_with_args
def get_activities_for_users(user_ids: list, target_date: date, limit: int = 100):
"""여러 사용자의 활동을 동시 조회"""
parameters = [(uid, target_date, limit) for uid in user_ids]
results = execute_concurrent_with_args(
session,
select_stmt,
parameters,
concurrency=50,
raise_on_first_error=False
)
all_activities = []
for success, result in results:
if success:
all_activities.extend(list(result))
return all_activities
# Consistency Level 조정
def read_with_consistency(user_id: uuid.UUID, target_date: date):
"""높은 일관성 수준으로 읽기"""
stmt = SimpleStatement(
"SELECT * FROM user_activities WHERE user_id = %s AND activity_date = %s",
consistency_level=ConsistencyLevel.LOCAL_QUORUM
)
return session.execute(stmt, (user_id, target_date))
# 페이지네이션 (대용량 결과)
def get_all_activities_paged(user_id: uuid.UUID, target_date: date, page_size: int = 1000):
"""페이지네이션으로 대용량 데이터 조회"""
stmt = SimpleStatement(
"SELECT * FROM user_activities WHERE user_id = %s AND activity_date = %s",
fetch_size=page_size
)
result = session.execute(stmt, (user_id, target_date))
for row in result:
yield row
# result.has_more_pages로 더 많은 페이지 확인 가능
# 연결 종료
# cluster.shutdown()
DynamoDB 호환 API (Alternator)
import boto3
from botocore.config import Config
# ScyllaDB Alternator에 연결
dynamodb = boto3.resource(
'dynamodb',
endpoint_url='http://scylla-alternator:8000',
region_name='us-east-1',
aws_access_key_id='none',
aws_secret_access_key='none',
config=Config(
retries={'max_attempts': 3},
connect_timeout=5,
read_timeout=30
)
)
# 테이블 생성
table = dynamodb.create_table(
TableName='UserActivities',
KeySchema=[
{'AttributeName': 'pk', 'KeyType': 'HASH'},
{'AttributeName': 'sk', 'KeyType': 'RANGE'}
],
AttributeDefinitions=[
{'AttributeName': 'pk', 'AttributeType': 'S'},
{'AttributeName': 'sk', 'AttributeType': 'S'}
],
BillingMode='PAY_PER_REQUEST'
)
# 아이템 삽입
table = dynamodb.Table('UserActivities')
table.put_item(
Item={
'pk': 'USER#123',
'sk': 'ACTIVITY#2024-01-15T10:30:00',
'activity_type': 'login',
'details': {'ip': '192.168.1.1'}
}
)
# 쿼리 (DynamoDB API)
response = table.query(
KeyConditionExpression='pk = :pk AND begins_with(sk, :sk_prefix)',
ExpressionAttributeValues={
':pk': 'USER#123',
':sk_prefix': 'ACTIVITY#2024-01'
}
)