Dynamo

From Big Data Analytics Lab

Dynamo: Amazon’s Highly Available Key-value Store

Giuseppe DeCandia, Deniz Hastorun, Madan Jampani, Gunavardhan Kakulapati, Avinash Lakshman, Alex Pilchin, Swaminathan Sivasubramanian, Peter Vosshall, Werner Vogels: Dynamo: amazon's highly available key-value store. SOSP 2007: 205-220

Background

  • RDBMS를 사용할 때의 문제점
    • 과도한 기능 -> 비싼 H/W와 운영 인력 필요 -> 비용 증가
      • 아마존에서는 많은 서비스들이 DB에 접근할 때 단순히 primary-key access만 하고, 복잡한 query 같은 건 없더라..
        • e.g., best seller lists, shopping carts, customer preferences, session management, sales rank, and product catalog
    • availability에 한계가 있음. 대신 consistency를 선택한 결과.
      • 기업의 입장에서는 서비스가 잠깐이라도 멈추면 막대한 영업 손실.
    • scale-out이 힘들고, load balancing을 위해 잘 partitioning하는 것도 어려움.
      • 계속 늘어가는 데이터를 처리할 수 있어야 함.

System Assumptions and Requirements

  • Query model
    • primary-key로 접근하는 단순 read/write.
    • value는 비교적 작은 binary object(즉, blob). 대략 1MB 이하 가정.
    • 한 data item에 대한 operation만 지원.
    • relational schema 필요 없음.
  • ACID 성질
    • high availability를 위해 consistency 완화.
    • isolation 보장 안함.
    • atomicity는 single key update만 허용.
  • Efficiency
    • 철저히 low latency를 보장해야 함.
      • peak time에 초당 어느 정도 request가 들어올 때 99.9% 이상에 대해 일정 응답시간 이하 보장.
  • Other assumptions
    • 아마존 내부에서만 쓰이므로 보안 관련 요구사항은 없음.
    • 수백 노드 이상 scalability 제공.

Design Considerations

Consistency

  • RDBMS: synchronous replica coordination을 해서 강한 consistency를 보장하지만 availabilty는 취약.
  • Dynamo: optimistic replication 기법을 사용해서 availability를 높임. concurrent, disconnected work에 대해서는 괜찮은 수준으로만 처리.
    • eventual consistency: 모든 update는 종내 모든 replica에 도달한다.

Resolving Update Conflict

언제 할 것인가

  • RDBMS: write할 때 처리. 모든 replica에 쓰지 못하면 write 실패.
    • RDBMS에서는 common case인 read를 빠르게 하는 것이 중요.
  • Dynamo: "always writable". read할 때 처리.
    • 아마존에서는 고객의 update를 거부하는 것은 고객에게 나쁜 체험을 안겨줌.
      • e.g., 쇼핑 카트에 아이템을 담거나 삭제하는데 실패하는 경우

누가 할 것인가

  • data store에서 처리: 단순한 policy만 쓸수 있음. "last write wins"
  • application에서 처리: 각 app에 맞는 방법을 쓸수 있음.
    • e.g., 쇼핑 카트에서 update 충돌 시 서로 merge해서 처리.
  • app에서 처리하지 않으면 data store 단에서 처리하도록 함.

기타 주요 원칙

  • incremental scalability: 노드 추가를 쉽게.
  • symmetry & decentralization: 모든 노드는 동일한 책임을 지님. 특별한 노드가 없음. 중앙 제어 방식 배제.
  • heterogeneity: 각 서버의 성능에 맞게 작업 분배.

System Architecture

Dynamo에서 사용한 기법들 및 장점 요약
Problem Technique Advantage
Data Partitioning Consistent Hashing Incremental Scalability
High Availability for writes Vector clocks with reconciliation during reads Version size is decoupled from update rates
Handling temporary failures Sloppy Quorum and hinted handoff Provides high availability and durability guarantee when some of the replicas are not available
Recovering from permanent failures Anti-entropy using Merkle trees Synchronizes divergent replicas in the background
Membership and failure detection Gossip-based membership protocol and failure detection Preserves symmetry and avoids having a centralized registry for storing membership and node liveness information

Data Processing

System Interface

단순 key access를 위해 딱 두가지 operation만 제공: get(), put()

  • get(key): 하나의 object / conflict 시 여러 버전 object의 리스트 + context
  • put(key, context, object): context는 version 관리를 위한 정보 등의 system metadata를 담음.
    • insert, delete 모두 put()으로 처리됨.
  • key와 object 모두 opaque byte array로 간주. key에는 MD5 hash 사용.

Partitioning

Consistent Hashing

원래 hashing은 노드 개수(hash slot 개수)가 바뀌면 모든 key를 다시 mapping해야 하는데 이를 평균 k/n개만 바꾸면 되도록 개선. (k: 키 개수, n: 노드 개수)

  • hash 값의 range를 어떤 고정된 circular space("ring")의 위치로 함. (즉, angle 값.) 노드들을 ring에 random하게 분포시키고 각 key를 hash한 위치에서 시계방향으로 걸어가서 나오는 첫 노드에게 할당하는 방식.
  • 노드가 추가/제거되었을 때 그 노드의 바로 다음 노드만 영향을 받고, 나머지는 그대로임.
Partitioning and replication of keys in Dynamo ring

Dynamo가 사용하는 변종 consistent hashing

노드들의 성능이 다른 경우 및 load balancing을 고려하여 consistent hashing을 개선.

  • "virtual nodes 개념 도입: 한 노드에 ring의 여러 위치 할당 가능.
  • 장점
    • 한 노드가 죽었을 때 그로 인한 load가 여러 노드로 분산됨.
    • 한 노드가 추가될 때 여러 노드로부터 load를 골고루 가져옴.
    • 노드의 성능에 따라 virtual node 할당 개수를 조정함으로써 효율적 자원 이용 가능.

Replication

  • 위의 consistent hashing으로 할당된 노드가 coordinator node가 되어 replication을 책임짐.
  • 자신의 local에 저장하고, 시계 방향으로 N-1개의 노드에 replication하도록 함.
    • virtual node 개념 때문에 N개 노드 중에 물리적으로 같은 노드가 들어 있으면 실제로 N 노드에 저장이 안됨. 이를 고려하여 물리적으로 같은 노드는 skip하면서 저장할 노드들(특정 key의 preference list라고 함)을 선택.
    • 모든 노드가 어떤 key에 대해서든 preferece list를 판단할 수 있도록 설계됨.
    • 여러 data center에 replica가 골고루 분산될수 있도록 preference list를 만들어야 함.

Data Versioning

Data가 여러 버전이 생기는 이유

  • Dynamo는 eventual consistency만을 제공하며, 모든 replica가 asynchronous하게 propagation됨.
    • update가 모든 replica에 전달되지 않은 상황에서 뒤이어 바로 read가 들어온 경우 최신 update가 반영되지 않은 예전 object가 return될 수 있음.
    • server outage나 network partitions 등의 failure scenario에서는 update가 모든 replica에 전달되는 시간이 길어질 수도 있음.
  • 아마존에는 이러한 정도의 inconsistency를 감내할 만한 application들이 많이 있음.
    • e.g., shopping cart - add to cart / delete item from cart
  • Dynamo는 각 modification의 결과물을 new & immutable 버전으로 처리. 한 object에 대해 동시에 여러 버전이 있을 수 있음.
    • network partition, node failure 등으로 인해 동일한 데이터의 버전이 두 개뿐만 아니라 여러 개일 수도 있음.

Conflict Resolution

  • 새 버전이 이전 버전(들)을 포함하는 경우: 시스템이 새 버전을 authoritative 버전으로 결정. (syntactic reconciliation)
  • 버전들이 branch가 발생한 경우: client가 reconciliation을 해주어야 함. (semantic reconciliation)
    • e.g., shopping cart - "add to cart" 결과는 절대 사라지지 않음. 반면에 삭제했던 item은 다시 나타날 수도 있음. 사용자가 merge할 수 있도록.
  • 버전들의 인과 관계를 알아내기 위해서는 vector clock 이용.

Maintaining Consistency

  • quorum system과 유사한 consistency protocol 사용.
  • latency를 좋게 하기 위해 보통 R + W < N 이 되도록 설정. (R: read / W: write operation이 성공한 걸로 간주하기 위해 필요한 최소 노드 수)

Cluster Management

Membership and Failure Detection

Ring Membership

  • ring에서 완전히 노드 추가/제거하는 건 관리자가 명시적으로 처리.
    • 드물기 때문. 따라서 replica들을 rebalancing하는 일도 잘 발생하지 않음.
    • 실수로 노드를 추가하는 일도 막을 수 있음.
  • gossip-based protocol 사용해서 membership의 변화를 모든 노드에 전파.
    • 매 초마다 random하게 고른 peer와 통신해서 membership 변경 기록을 reconcile.

External Discovery

  • 일부 노드들끼리만 통신하여 logical network partition 문제가 생기는 것을 막기 위해 몇몇 노드를 seed로 지정.
    • seed는 모든 노드에게 알려짐. 명시적으로 주소를 설정.

Failure Detection

  • temporary node failure: 개별 노드들에 의해 감지됨. 메시지에 응답이 없는 경우.
  • permanent node failure: 관리자가 명시적으로 join/leave method 수행.

Adding/Removing Storage Nodes

  • 노드 추가/제거 시에는 할당된 key range에 따라 N개의 노드들로부터 key들을 넘겨받거나 넘겨줌.

Handling Failures

Handling Temporary Failures

  • Sloppy quorum
    • 모든 read/write operation은 preference list 중에 처음 N개의 healthy 노드에 수행된다. 항상 처음 N개가 아님에 유의.
    • high availability를 위해 quorum membership을 유연하게 가져감.
  • Hinted handoff
    • write operation 중 어떤 노드가 일시적으로 다운됐거나 네트웍이 연결이 안되는 경우, preference list의 순서에 따라 이를 대신 수행하는 노드가 생김.
    • 이러한 경우 replica를 저장할 때 metadata에 원래 저장했어야 하는 노드를 hint로 적어둠.
    • hint가 달린 replica들은 따로 local db에 두고 주기적으로 스캔. 원래의 노드가 살아난 경우 replica를 그리로 보내주고 자신은 제거.

Handling Permanent Failures

hinted replica를 가진 노드가 original replica 노드가 살아나기 전에 죽는 등의 시나리오에서 오래된 replica들이 계속 남아 있을 수 있다.
그에 따라 문제가 발생할 소지를 줄이고 durability를 높이기 위해 replica synchronization을 수행.
Merkle tree를 이용한 anti-entropy protocol 사용.

  • Merkle tree
    • leaf node는 개별 key 값의 value들의 hash 값.
    • parent node는 child 노드들의 값들을 concatenation해서 다시 hash한 값.
    • 두 Merkle tree의 노드 값이 동일면 child 노드들은 검사해보지 않아도 동일함을 알 수 있다.
  • Dynamo에서 Merkle tree를 사용하는 방식
    • 각 노드마다 각 key range(한 virtual node가 담당하는 key들의 집합) 마다 하나의 Merkle tree를 유지.
    • 두 노드가 root의 hash 값부터 교환하면서 바뀐 부분을 찾아서 replica synchronization.
    • 단점: 시스템에서 노드가 추가/삭제되면 key range가 바뀌어서 tree 재생성이 필요하다.

Implementation

Local Persistence Component

  • 여러 종류의 storage engine을 plug-in해서 사용할 수 있음.
  • 다루는 object의 크기에 따라 선택해서 사용.
    • Berkeley Database (BDB) Transactional Data Store: 제일 많이 사용. 다루는 object의 크기가 수십 KB 정도인 경우
    • MySQL: object 크기가 더 클때 사용.
    • BDB Java Edition, in-memory buffer with persistent backing store 등

Request Coordination Component

  • 모든 통신은 Java NIO 채널을 사용해서 구현.
  • 하나의 client request마다 요청을 받은 노드에 state machine instance를 생성. 요청을 책임짐.
    • key에 대한 책임이 있는 노드 확인
    • 노드들에게 request 보내기, response 기다리기
    • 필요한 경우 retry하기
    • reply 처리해서 client에게 packaging
    • read repair: client에게 reply한 후에도 다른 노드들의 'outstanding' response를 받기 위해 약간동안 더 기다림.
      • 혹시 stale version의 데이터가 오면 fresh version을 보내줘서 업데이트하도록 시킴. replica sync.의 부담을 덜도록.

Experience & Lessons Learned

  • application에 맞게 version reconciliation logic 및 quorum parameter 설정을 조정 가능.
    • business logic specific reconciliation: 서비스에 맞는 고유 logic 사용.
      • e.g., 쇼핑 카트 - 다른 버전을 merging
    • timestamp based reconciliation: "last write wins"
      • e.g, 고객 세션 정보 - 가장 큰 timestamp의 것만 남김
    • high performance read engine: write가 적고 read만 많은 경우
      • e.g., 상품 카탈로그, 광고 아이템 - R = 1, W = N 으로 quorum parameter 조정.
Partitioning and placement of keys in the three strategies
  • 메모리에 object buffer를 두고 write를 모아서 background thread가 주기적으로 처리.
    • durability를 낮추고 performance를 높임
  • consistent hashing의 partitioning & placement 최적화
    • strategy 1: T random tokens per node and partition by token value (앞에 나온 내용)
      • token이 random하게 선택되고, range의 크기가 다양함.
      • hand-off 시 문제가 있음
        • 새 노드 join시 key range를 넘겨주어야 하는데 넘겨줄 item을 찾기 위해 local persistence store를 scan해야 함.
        • 바뀐 range에 맞게 Merkle tree도 다시 계산해야 함.
      • 전체 key space의 snapshot을 얻는 과정에서 key range가 다양해서 archive가 어려움 (?)
      • 이러한 문제들은 기본적으로 data partitioning과 placement가 밀접하게 엮여 있는데서 기인.
    • strategy 2: T random tokens per node and equal sized partition
      • 파티션은 동일한 크기로 하자... strategy 1->3으로 가는 중간단계.
    • strategy 3: Q/S tokens per node, equal-sized partition
      • 동일한 크기의 파티션(Q: 파티션 개수) + 파티션 사이 사이 지점을 token으로 두고 노드마다 Q/S개의 token 할당(S: 노드 개수)
        • 장점: strategy 1의 문제 해결. bootstrap, recovery가 빨라짐. archive가 쉬움.
        • 단점: node membership이 바뀔 때 coordinate 필요.
  • data item이 여러 버전이 발생하는 이유: failure (node, data center, network), concurrent write
    • 24시간 통계 결과 99.94%의 경우에 1개 버전만 리턴.
  • server의 load balancer를 쓰지 않고 client-driven coordination을 사용하면 latency가 좋아짐.
    • client가 주기적으로 노드 membership 정보를 업데이트하고 이를 이용해서 request 처리.
  • background task(replica sync, data handoff 등) 는 admission control 메커니즘을 통해 foreground task(get, put 처리) 에 큰 영향을 주지 않을 정도로만 동작.
    • admission controller가 99.9% 서비스 수준 요구사항을 지키기 위해 꾸준히 latency 등의 상태 정보를 monitoring하고 그에 따라 time slice를 bg process에게 할당.