분산 아키텍처: Cassandra는 마스터-슬레이브 구조 없이 모든 노드가 동등한 피어(Peer-to-Peer) 방식으로 동작합니다. 링(Ring) 구조로 데이터를 분산 저장하며, 일관된 해싱(Consistent Hashing)을 통해 파티션 키 기반으로 데이터를 분배합니다. 단일 장애점(SPOF)이 없어 고가용성을 보장합니다.
데이터 모델: Wide Column Store 모델을 사용합니다. 키스페이스(Keyspace) > 테이블 > 파티션 > 로우 > 컬럼 구조로 데이터를 저장합니다. 파티션 키는 데이터 분산의 핵심이며, 클러스터링 키로 파티션 내 정렬 순서를 정의합니다. CQL(Cassandra Query Language)로 SQL과 유사하게 쿼리할 수 있습니다.
일관성 레벨: 튜너블 일관성(Tunable Consistency)을 지원합니다. ONE, QUORUM, ALL 등 다양한 일관성 레벨을 쿼리별로 설정할 수 있어, 일관성과 가용성 사이의 트레이드오프를 조절할 수 있습니다. 쓰기는 빠르게, 읽기 시 복제본을 비교하는 Read Repair로 최종 일관성을 보장합니다.
복제와 내결함성: 복제 계수(Replication Factor)를 설정하여 여러 노드에 데이터를 복제합니다. NetworkTopologyStrategy로 데이터센터별 복제 전략을 다르게 설정할 수 있어, 지역 간 재해 복구(DR)에 효과적입니다. 힌트 핸드오프(Hinted Handoff)로 일시적 장애 시에도 데이터 손실을 방지합니다.
주요 사용 사례: Netflix(스트리밍 메타데이터), Apple(10만+ 노드 클러스터), Instagram(메시지 저장), Discord(채팅 메시지), IoT 센서 데이터, 시계열 분석, 로그 저장 등 대규모 쓰기 워크로드에 적합합니다.
💻 코드 예제
# Cassandra Python 예제 - cassandra-driver
from cassandra.cluster import Cluster
from cassandra.auth import PlainTextAuthProvider
from cassandra.policies import DCAwareRoundRobinPolicy
from cassandra.query import SimpleStatement, ConsistencyLevel
from datetime import datetime
import uuid
class CassandraService:
"""Cassandra 데이터베이스 서비스"""
def __init__(self, contact_points: list, keyspace: str):
# 인증 설정 (선택적)
auth_provider = PlainTextAuthProvider(
username='cassandra',
password='cassandra'
)
# 클러스터 연결 (로드 밸런싱 정책 포함)
self.cluster = Cluster(
contact_points=contact_points,
auth_provider=auth_provider,
load_balancing_policy=DCAwareRoundRobinPolicy(local_dc='datacenter1')
)
self.session = self.cluster.connect(keyspace)
def create_schema(self):
"""키스페이스 및 테이블 생성"""
# 키스페이스 생성 (복제 계수 3)
self.session.execute("""
CREATE KEYSPACE IF NOT EXISTS iot_data
WITH replication = {
'class': 'NetworkTopologyStrategy',
'datacenter1': 3
}
""")
# 시계열 테이블 생성 (파티션 키 + 클러스터링 키)
self.session.execute("""
CREATE TABLE IF NOT EXISTS iot_data.sensor_readings (
device_id UUID,
reading_date DATE,
reading_time TIMESTAMP,
temperature FLOAT,
humidity FLOAT,
PRIMARY KEY ((device_id, reading_date), reading_time)
) WITH CLUSTERING ORDER BY (reading_time DESC)
""")
print("스키마 생성 완료")
def insert_sensor_data(self, device_id: uuid.UUID, temp: float, humidity: float):
"""센서 데이터 삽입 (배치 처리)"""
now = datetime.now()
# Prepared Statement 사용 (성능 향상)
insert_stmt = self.session.prepare("""
INSERT INTO sensor_readings
(device_id, reading_date, reading_time, temperature, humidity)
VALUES (?, ?, ?, ?, ?)
""")
# 일관성 레벨 설정
bound_stmt = insert_stmt.bind([
device_id,
now.date(),
now,
temp,
humidity
])
bound_stmt.consistency_level = ConsistencyLevel.QUORUM
self.session.execute(bound_stmt)
print(f"데이터 삽입: device={device_id}, temp={temp}")
def get_recent_readings(self, device_id: uuid.UUID, date, limit: int = 100):
"""최근 센서 데이터 조회"""
query = SimpleStatement(
"""
SELECT reading_time, temperature, humidity
FROM sensor_readings
WHERE device_id = %s AND reading_date = %s
LIMIT %s
""",
consistency_level=ConsistencyLevel.LOCAL_QUORUM
)
rows = self.session.execute(query, [device_id, date, limit])
return list(rows)
def batch_insert(self, readings: list):
"""배치 삽입 (대량 데이터)"""
from cassandra.query import BatchStatement, BatchType
batch = BatchStatement(batch_type=BatchType.UNLOGGED)
insert_stmt = self.session.prepare("""
INSERT INTO sensor_readings
(device_id, reading_date, reading_time, temperature, humidity)
VALUES (?, ?, ?, ?, ?)
""")
for reading in readings:
batch.add(insert_stmt, reading)
self.session.execute(batch)
print(f"{len(readings)}건 배치 삽입 완료")
def close(self):
"""연결 종료"""
self.cluster.shutdown()
# 사용 예시
if __name__ == "__main__":
# 클러스터 연결
cassandra = CassandraService(
contact_points=['127.0.0.1', '127.0.0.2', '127.0.0.3'],
keyspace='iot_data'
)
# 스키마 생성
cassandra.create_schema()
# 데이터 삽입
device_id = uuid.uuid4()
cassandra.insert_sensor_data(device_id, 25.5, 60.0)
# 데이터 조회
from datetime import date
readings = cassandra.get_recent_readings(device_id, date.today())
for row in readings:
print(f"시간: {row.reading_time}, 온도: {row.temperature}°C")
cassandra.close()