[mongodb] Sharding

sharding은 mongodb의 scaling out을 제공한다.
sharding은 db를 사용하는 application의 영향없이,
load와 data size 증가 처리를 도와준다.

* sharding 소개

sharding은 서로 다른 machine에 data를
쪼개고(splitting) 다른 부분들을 저장하는 process를 말한다.
partitioning 용어도 쓰이기도 한다.
machine들간에 data를 쪼개는것에 의해,
성능 좋고 대용량의 machine 없이도 더 많은 data와 그 처리를 가능토록 한다.

수동(manual) sharding은 대부분의 db software에서 제공된다.
application에서 몇개의 다른 server로 접속을 관리할 때,
각각은 서로 독립적으로 된다. application은 다른 server에 다른 data를
저장하도록 하며, 적합한 server에 data를 구하도록 query한다.
이러한 application은 잘 동작하지만, cluster를 추가/삭제하는데
어렵고, load 패턴이나 data의 분산정도가 변경될때도 마찬가지 이다.

mongodb는 autosharding을 지원하는데, 이는,
관리가 어려웠던 수동 sharding의 고통을 제거한다.
cluster는 data를 쪼개는 작업을 처리하고, 자동으로 re-balancing 한다.
이제부터 sharding와 autosharding은 같은 뜻으로 고려한다.
물론 manual sharding과는 차이가 있다.

* mongodb의 autosharding

mongodb autosharding의 기본 개념은 collection을 더작은 chunk로 나누는
것이다. 이러한 chunk들은 shard들을 경유해 분배되는데, 이는 각각의 shard가
total data set의 부분 집합으로 책임을 지기 위해서이다. 우리는 어떤 shard가
어떤 data를 가지고 있는지, 혹은 data가 다수의 shard에 어떻게 쪼개졌는지
알고 싶지는 않다. 그래서 shard에 앞서 mongos라고 불리는 routing process를
실행시키는 것이다. 이러한 routerr는 모든 data의 위치를 알고 있으며,
application은 그것에 접소하고 일반적인 request를 수행한다. application에서
알기로는, 일반 mongod에 connect된다. router는 어떤 data가 어떤 shard에 있는지를
알고 있고, 그래서 적합한 shard들에 request를 전송한다. 만일 request에 응답이 있다면,
router는 그들을 수집하고 application에 전달한다.

nonshard mongodb는 아래와 같이 client는 mongod process에 접속한다.

 client <--> mongod

shard된 경우는 아래와 같이 mongos process에 접속한다.

client <--> mongos <---> mongod
client <--> mongos <-|
client <--> mongos <-|-> mongod
client <--> mongos <-|
client <--> mongos <---> mongod

shard는 언제?

"언제 sharding을 시작해야 하는가?"로 종종 질문을 받는다.
"sharding이 좋은 생각이다~!"라는 징조는 다음과 같다.

- 현재 machine에 disk space가 떨어졌을 때
- 단일 mongod가 처리할 수 있는 속도보다 빨리 data를 쓰고 싶을 때
- 성능 향상을 위해 memory에 있는 data의 크기를 키우고 싶을 때

일반적으로, nonshard로 시작하고, 필요시 그것을 shard로 변경한다.

* sharding의 key

sharding이 이뤄졌다면, collection으로 부터 key를 선택해야 하며,
그 값으로 data를 쪼개게 된다. 이러한 key를 shard key라 한다.

예를 들어 보자.
people을 표현하는 document의 collection이 있다고 가정하자.
shard key로 name을 선택했다면, 하나의 shard는 A~F까지로 시작하는 "name"을
가지고 있을 것이다. 물론 그다음 shard는 G~P일 것이고, 마지막으로 Q~Z일 것이다.
만일 shard를 추가(혹은 삭제)했을 때, mongodb는 이러한 data를 기반으로 re-balance를
실행하도록 한다.
(만일 traffic이 높은 임의의 shard는, 덜 사용되는 다른 shard보다 더 작은양의 data가 들어갈 것이다)

기존 collection의 sharding

기존에 있는 log 용도의 collection이 있고, 그것을 sharding해 보자.
만일, shard key로 "timestamp"를 사용한다면, 모든 data에 대해 단일 shard로 구성된다.
어떤 data를 insert하면, mongodb는 하나의 shard로 할 것이다.

이제, 새로운 shard를 추가한다고 가정하자.
추가된 shard가 이제 동작중이라면, mongodb는 2개의 chunk로 collection을 쪼갠다.
chunk는 shard key에 대한 값의 범위에 대한 정보를 가지고 있다.
그리고, -무한대~2003/07/26, 그리고 2003/07/27~+무한대와 같이 timestamp기준으로
쪼개진다. 이러한 chunk들중 하나는 다른 shard로 옮겨갈 것이다.

2003/07/27 이전의 timestamp를 가지는 document가 추가되면,
첫 chunk에 추가된다. 그렇지 않다면 두번째 chunk에 저장된다.

자연 증가(incrementing) shard key vs. 랜덤 shard key

end user 입장에서 shard 설치는 nonshard된 것과 구별이 되어서는 안된다.
그러나, shard key가 선택되었느냐에 따라 query가 다르다는것을 이해하면,
유용할 때가 만다.

앞서, "name" key에 대해 shard된 경우를 예를 들어 보자. 물론, 3개의 shard이다.

db.people.find({"name":"Susan"})
; mongos는 바로 Q-Z shard에 query할 것이다. 해당 shard로 부터 수신되면 client에 보낸다.

db.people.find({"name":{"$lt":"L"}})
; mongos는 A-F, G-P shard들에게 연속으로 query한다. 그들의 응답으로 client에 보낸다.

db.people.find().sort({"email":1})
; mongos는 모든 shard에 대해 query하고, 그 결과를 merge sort한다.
mongos는 각 server로 부터의 받은 cursor를 사용하며, 그래서 전체 dataset을 구할
필요는 없다.

db.people.find({"email":"joe@example.com"})
; mongos는 email key의 발자취를 가지고 있지 않다. 그래서,
어떤 shard가 보내져야 하는지를 모른다. 그래서 연속으로 모든 shard에 query를 보낸다.

만일, 새로운 document가 insert된다면, mongos는 적당한 shard로 "name" key의 값을 기반으로
보낸다.

sharding 세팅하기

크게 다음과 같이 2개의 과정을 거친다.

- server 시작하기
- data를 어떻게 shard할지 결정

sharding은 기본으로 다음의 서로 다른 component들이 함께 작업을 진행한다.

shard
; collection data의 부분 집합을 가지는 container.
shard는 단일 mongod server 혹은 replica set이다.
따라서, shard에 많은 server가 있더라도 master가 하나라면 모든 server는 동일 data를 가진다.

mongos
; router process이다. 기본적으로 request를 route시킨다. 그리고 결과를 종합한다.
그리고 어떠한 data나 설정 정보를 저장하지 않는다.
(config server로 부터 정보를 cache는 한다)

config server
; cluster 설정을 저장하는 server이다.
설정이란, 어떤 shard에 어떤 data를 저장하냐? 이다.
mongos는 영구적으로 저장하지 않기 때문에, shard 정보를 가져오기 위해서 필요했다.
config server로 부터 data를 동기화한다.

이미 mongodb 작업을 하고 있었다면, shard를 준비할 수 있다.
(현재의 mongod는 당신의 첫 shard일 것이다)
다음은 부담없이 새로운 shard를 생성하는 과정을 보여준다.

server 시작하기

처음으로 config와 mongos server를 시작해야 한다.
config server가 먼저 시작되어야 한다.
config server는 다음과 같이 mongod와 비슷한 방식으로 시작할 수 있다.

$ mkdir -p ~/dbs/config
$ mongod --dbpath ~/dbs/config --port 20000

config server는 많은 disk space나 시스템 자원을 필요하지는 않는다.
(대략 200MB의 실제 data당 1KB 정도의 크기를 예상할 수 있다.)

이제 application에서 접속할 mongos process가 필요하다.
routing server는 data directory를 필요하지 않고, config server의 위치가
필요할 뿐이다.

$ mongos --port 30000 --configdb localhost:20000

shard 추가하기
; shard는 mongod instance이다.(혹은 replica set)
$ mkdir -p ~/dbs/shard1
$ mongod --dbpath ~/dbs/shard1 --port 10000

이제 mongos process에 접속하고 cluster에 shard를 추가한다.
mongos에 아래와 같이 shell로 접속한다.
$ mongo localhost:30000/admin

db.runCommand({addshard:"localhost:10000", allowLocal:true})
{
    "added" : "localhost:10000",
    "ok" : true
}

allowLocal은 localhost에 shard가 동작하고 있는 경우에만 필요하다.
mongodb는 cluster를 local에 저장하는 것을 원하진 않는다.
만일 릴리즈를 할 경우에는, 반드시 shard를 다른 machine으로 해야 한다.

data sharding하기

mongodb는 저장된 data의 모든 조각을 분산하진 않는다.
명시적으로 database와 collection level로 sharding해야만 가능하다.

다음과 같은 예를 고려해보자.
foo database의 "_id" key의 bar collection을 shard한다.
우선, foo에 대해 sharding 시키자.

db.runCommand({"enablesharding":"foo"})

collection의 database sharding 결과는 각각의 다른 shard에 저장된다.

일단 database level에서 sharding되었다면, shardcollection 명령으로
collection으로 shard할 수 있다.

db.runCommand({"shardcollection":"foo.bar", "key":{"_id":1}})

해당 collection은 "_id" key에 의해 shard된다.
data를 추가하게 되면, _id 값을 기반으로 자동으로 분산하게 된다.

* Production Configuration

만일 production으로 application을 이동할 경우, 보다 견고한 setup이 필요하다.
실패없이 sharding하기 위해서 다음이 필요하다.

- 다중 config server
- 다중 mongos server
- 각 shard 당 replica set
- w 세팅 ([mongodb] Replication으l w 참고)

견고한 config

다중 config server는 간단하다.
만일 한개의 config server(개발용) 혹은 3개의 config server(production용)가 있다고 하자.

다중 config server는 다 동일하다. 즉, 3번 실행하면 된다.

mkdir -p ~/dbs/config1 ~/dbs/config2 ~/dbs/config3
mongod --dbpath ~/dbs/config1 --port 20001
mongod --dbpath ~/dbs/config2 --port 20002
mongod --dbpath ~/dbs/config3 --port 20003

mongos를 시작할때는, 다음과 같다.

mongos --configdb localhost:20001,localhost:20002,localhost:20003

config server는 두단계 commit을 사용한다.
cluster configuration의 개별 복사본을 유지하는데,
일반적인 mongodb 비동기 복사를 사용하지 않는다.
이는 단일 configuration server가 down되면,
cluster의 configuration 정보는 read-only가 된다.
client는 read와 write가 되는데, config server가 복원될 때 까지
re-balance는 이뤄지지 못한다.

많은 수의 mongos

많은 mongos를 원하는 만큼 운영할 수 있다.
하나의 추천된 setup은 모든 application server에 대해 mongos process를
실행시키는 것이다. 이는, 각각의 application server는 local의 mongos에
접근할 수 있으며, 그리고 만일 해당 server가 down 되더라도 그곳에 없는
mongos에 접근을 시도할자는 없을 것이다.

견고한 shard

production에서, 각 shard는 replica set으로 될 것이다.
이는 개별 server는 실패될 수 있으나, 전체 shard를 down 시키지는 않는다는 것이다.
shard로서 replica set을 추가할 때, 이름을 전달하고 addshard 명령을 사용한다.

db.runCommand({"addshard":"foo/prod.example.com:27017"})

만일, prod.example.com이 down된다면, mongos는 replica set에 연결되었음을 아고
새로운 primary를 이용하도록 한다.

물리적 server

아래는 압도할 만한 machine 수가 된다:
3개의 config server, shard당 최소 2개의 mongod, 많은 mongos process....
그러나, 모든 것이 고유의 machine을 가지고 있진 않다.
피해야할 사항중 하나가 하나의 machine으로 전체 component를 구성하는 일이다.
예를들어, 3개의 config server 혹은 mongos process 혹은 전체 replica set을
하나의 machine에 구성하는 일이다.
그러나, config server와 mongos process는 replica set의 member와 함게 공유 가능하다.

* sharding 관리하기

sharding 정보는 config database에 대부분 저장된다.
이는 mongos process에 접속된것으로 부터 접근된다.

config collections

다음 부터의 모든 code는 mongos process에 shell로 접속한 경우이며,
use config를 이미 마친 것으로 가정한다.

shards
; shard list를 구할 수 있다.
db.shards.find()
{ "_id" : "shard0", "host" : "localhosst:10000" }
{ "_id" : "shard1", "host" : "localhosst:10001" }

databases
; shard에 존재하는 database list와 정보를 구한다.

db.databases.find()
{ "_id" : "admin", "partitioned" : false, "primary" : "config" }
{ "_id" : "foo", "partitioned" : false, "primary" : "shard1" }
{ "_id" : "x", "partitioned" : false, "primary" : "shard0" }
{
    "_id" : "test",
    "partitioned" : true,
    "primary" : "shard0",
    "sharded" : {
        "test.foo" : {
            "key" : {"x" : 1},
            "unique" : false
        }
    }
}

 "_id" : string
 ; database의 이름

 "partitioned" : boolean
 ; enablesharding이 실행된 상태일때 true

 "primary" : string
 ; shard "_id"에 해당된다.

chunks
; chunk 정보는 chunks collection에 저장된다.
data가 cluster 상에 어떻게 쪼개졌는지 보여준다.

db.chunks.find()
{
    "_id" : "test.foo-x_MinKey",
    "lastmod" : { "t" : 1276636243000, "i" : 1 },
    "ns" : "test.foo",
    "min" : {
        "x" : { $minKey : 1 }
    },
    "max" : {
        "x" : { $maxKey : 1 }
    },
    "shard" : "shard0"
}

chunk range는 -무한대(MinKey)에서 +무한대(MaxKey)가 된다.

sharding 명령

summary 구하기

printShardingStatus는 이전 collection의 quick summary를 제공한다.

db.printShardingStatus()

--- Sharding Status ---
  sharding version: { "_id" : 1, "version" : 3 }
  shards:
      { "_id" : "shard0", "host" : "localhost:10000" }
      { "_id" : "shard1", "host" : "localhost:10001" }
  databases:
      { "_id" : "admin", "partitioned" : false, "primary" : "config" }
      { "_id" : "foo", "partitioned" : false, "primary" : "shard1" }
      { "_id" : "x", "partitioned" : false, "primary" : "shard0" }
      { "_id" : "test", "partitioned" : true, "primary" : "shard0",
          "sharded" : { "test.foo" : { "key" : { "x" : 1 }, "unique" : false } } }
               test.foo chunks:
                   { "x" : { $minKey : 1 } } -->> { "x" : { $maxKey : 1 } } on : shard0
                   { "t" : 1276636243000, "i" : 1 }

shard 제거하기

db.runCommand({"removeshard" : "localhost:10000"});
와 같이 한다.