하둡 완벽 가이드 - 21. 주키퍼

사용자의 입장으로만 하둡을 바라보게 되어 깊이가 부족하다는 생각을 하게 되었다.
하둡 완벽 가이드를 읽고 이해한대로 정리한다.

---

주키퍼는 하둡의 분산 코디네이션 서비스로, 분산 애플리케이션을 구축할 수 있다.
분산 애플리케이션의 부분 실패(partial failure)가 가장 큰 분산 애플리케이션 작성에 어려움이다. 
네트워크로 연결된 두 노드 사이에 메시지가 전송된 후 네트워크가 끊긴 경우의 상황이다. 100% 피할 수는 없어서 주키퍼를 사용한다고 해도 부분 실패가 완전히 사라지는 것은 아니며 완벽히 감출 수도 없다. 그러나, 안전하게 처리할 수 있다.

쥬키퍼의 특징

* 단순하다
* 제공하는 기능이 풍부하다
* 고가용성을 제공한다.
* 느슨하게 연결된 상호작용에 도움을 준다.
* 라이브러리다.
* 쓰기 위주의 작업 부하 벤치마크에서 초당 10,000개 이상, 읽기 위주에서는 그보다 몇 배 높은 처리량을 보인다.

21.1 주키퍼 설치와 실행

java 설치하고, 주키퍼 릴리즈 버전 내려받아 압축 풀고 환경 설정 파일 구성한다. 그리곤 zkServer.sh start로 시작!

주키퍼는 다양한 4글자 단어의 관리 명령어가 존재한다.

21.2 예제

어떤 외부 이벤트가 발생했을 때 엔트리의 상태를 능동적으로 변경하는 기능에 대해 설펴보자.

21.2.1 주키퍼의 그룹 멤버십

주키퍼는 파일과 디렉터리를 통합한 znode라 불리는 노드를 제공한다. znode는 데이터 컨테이너(파일에 해당)와 znode 컨테이너(디렉터리에 해당)의 역할을 둘 다 맡고 있다.

따라서 루트(/) 노드를 시작으로 계층적인 네임스페이스를 형성한다. 아래 이미지에서는 znode에 데이터를 저장하는 것이 보이지 않지만 실제로는 호스트명이나 IP주소와 같은 멤버 정보가 각각의 znode에 저장된다.

21.2.2 그룹 생성

Wacher를 붙여서 주키퍼에 그룹 znode를 생성하는 프로그램은 다음과 같다.

public class CreateGroup implements Watcher { 
  private static final int SESSION_TIMEOUT = 5000;
  private ZooKeeper zk;
  private CountDownLatch connectedSignal = new CountDownLatch(1);
  public void connect(String hosts) throws IOException, InterruptedException { 
    zk = new ZooKeeper(hosts, SESSION_TIMEOUT, this);
    connectedSignal.await();
  }
  @Override
  public void process(WatchedEvent event) { // Watcher interface
    if (event.getState() == KeeperState.SyncConnected) {
      connectedSignal.countDown();
    }
  }
  public void create(String groupName) throws KeeperException, InterruptedException {
    String path = "/" + groupName;
    String createdPath = zk.create(path, null/*data*/, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    System.out.println("Created " + createdPath);
  }
  public void close() throws InterruptedException {
    zk.close();
  }
  public static void main(String[] args) throws Exception { 
    CreateGroup createGroup = new CreateGroup();
    createGroup.connect(args[0]);
    createGroup.create(args[1]);
    createGroup.close();
  }
}

CreateGroup은 Wacher를 구현하여 클라이언트가 주키퍼에 연결되면 Watcher는 process() 메서드를 실행한다. 이 코드에서 Wacher는 CountDownLatch에 있는 카운터를 감소시키고 해당 숫자가 모두 종료되어야 다음으로 진행되도록 되어있으므로, connect 함수가 진행되기 위해서는 connectedSignal이 0이 되어야 한다. 

간단히,
connect 메서드 내부에서 주키퍼 연결 시도 -> await 으로 대기 -> 주키퍼 연결되면 process() 메서드 호출 -> create() 함수 실행

create() 메서드에서 호출하는 zk.create의 마지막 매개변수는 znode의 특성이다. 여기서는 영속 znode로 설정했고, 임시 znode도 설정할 수 있다. 임시 znode는 클라이언트 연결 해제시 주키퍼 서비스를 삭제되는 특성이 있다. 
이 프로그램에서는 znode 생성 후 클라이언트 연결이 종료되어도 유지하기를 원하므로 영속 znode로 설정했다.
create() 메서드의 반환값은 주키퍼가 생성한 경로이다. 

21.2.3 그룹 가입

위에서 만든 그룹에 멤버를 등록하는 프로그램을 만들어보자.

public class JoinGroup extends ConnectionWatcher {
  public void join(String groupName, String memberName) throws KeeperException, InterruptedException {
    String path = "/" + groupName + "/" + memberName;
    String createdPath = zk.create(path, null/*data*/, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
    System.out.println("Created " + createdPath);
  }
  public static void main(String[] args) throws Exception {
    JoinGroup joinGroup = new JoinGroup();
    joinGroup.connect(args[0]);
    joinGroup.join(args[1], args[2]);
    // stay alive until process is killed or thread is interrupted
    Thread.sleep(Long.MAX_VALUE);
  }
}

주키퍼와의 연결 성공을 대기하는 헬퍼 클래스

public class ConnectionWatcher implements Watcher {
  private static final int SESSION_TIMEOUT = 5000;
  protected ZooKeeper zk;
  private CountDownLatch connectedSignal = new CountDownLatch(1);
  public void connect(String hosts) throws IOException, InterruptedException {
    zk = new ZooKeeper(hosts, SESSION_TIMEOUT, this); 
    connectedSignal.await();
  }
  @Override
  public void process(WatchedEvent event) {
    if (event.getState() == KeeperState.SyncConnected) {
      connectedSignal.countDown();
    }
  }
  public void close() throws InterruptedException { 
    zk.close();
  }
}

Join Group 코드는 CreateGroup과 유사하다. join 메서드에서 znode를 생성할 때 EPHEMERAL으로 설정해서 프로세스가 잠들어있는게 강제로 종료되거나 시간만큼 잠들었다가 끝날때까지 기다렸다가 임시 znode가 자동으로 제거되는 것만 다르다.

21.2.4 그룹 멤버 목록

그룹의 멤버를 찾을 수 있는 프로그램은 다음과 같이 구현할 수 있다.

public class ListGroup extends ConnectionWatcher {
  public void list(String groupName) throws KeeperException, InterruptedException {
    String path = "/" + groupName;
    try {
      List<String> children = zk.getChildren(path, false);
      if (children.isEmpty()) {
        System.out.printf("No members in group %s\n", groupName);
        System.exit(1);
      }
      for (String child : children) { 
        System.out.println(child);
      }
    } catch (KeeperException.NoNodeException e) {
      System.out.printf("Group %s does not exist\n", groupName);
      System.exit(1);
    }
  }
  public static void main(String[] args) throws Exception {
    ListGroup listGroup = new ListGroup(); 
    listGroup.connect(args[0]);
    listGroup.list(args[1]);
    listGroup.close();
  }
}

쉽다. 그룹 이름이랑 getChildren() 메서드를 호출한다. 특정 znode에 감시를 설정하면 znode의 상태가 변할 때 등록된 Watcher에 신호가 전달되게 할 수도 있다.

이제 분산 시스템에 참여하는 노드 그룹의 목록을 구축하는 방법을 알았다. 노드는 서로 알 필요가 없고, 클라이언트의 존재 여부도 알 필요가 없다.

마지막으로, 그룹 멤버십 목록이 노드와의 통신 과정에서 발생하는 네트워크 에러를 처리할 수는 없다. 특정 노드가 그룹의 멤버라고 해도 그 노드와의 통신은 실패할 수 있다. 따라서 이러한 통신 장애는 통상적인 방법(재시도하거나, 그룹의 다른 멤버와 연결을 시도하거나..)으로 처리해야한다.

21.2.5 그룹 삭제

그룹과 모든 멤버를 삭제하는 코드는 다음과 같다. ZooKeeper 클래스는 경로와 버전 번호를 인자로 받는 delete() 메서드를 제공하므로 쉽게 그룹을 삭제할 수 있다. 버전번호를 -1으로 하면 버전과 상관없이 지워준다. 하지만, 재귀적 삭제가 안되므로, 부모 znode를 삭제하기 전에 자식 znode를 먼저 삭제해야한다.

public class DeleteGroup extends ConnectionWatcher {
  public void delete(String groupName) throws KeeperException, InterruptedException {
    String path = "/" + groupName;
    try {
      List<String> children = zk.getChildren(path, false); 
      for (String child : children) {
        zk.delete(path + "/" + child, -1);
      }
      zk.delete(path, -1);
    } catch (KeeperException.NoNodeException e) {
      System.out.printf("Group %s does not exist\n", groupName);
      System.exit(1);
    }
  }
  public static void main(String[] args) throws Exception { 
    DeleteGroup deleteGroup = new DeleteGroup(); 
    deleteGroup.connect(args[0]);
    deleteGroup.delete(args[1]);
    deleteGroup.close();
  }
}

21.3 주키퍼 서비스

주키퍼가 제공하는 데이터 모델, 기능, 구현과 같은 서비스의 특성을 살펴보자.

21.3.1 데이터 모델

주키퍼는 znode로 불리는 노드를 계층적 트리 형태로 관리한다. 각 znode는 데이터를 저장하고 연관 ACL을 가진다. 코디네이션 서비스를 위해 설계되었으므로, 대용량의 데이터를 저장하는 용도로는 사용할 수 없다. 1MB로 제한되어있다.

데이터 접근은 원자성(성공/실패)을 가진다. znode에 저장된 데이터를 읽을 때 일부만 받거나 일부만 갱신할 수는 없다. append도 없다. HDFS와 대조적이다.

절대경로만 지원한다. URI가 아니고 java.lang.String 클래스의 자바 API로 표현된다.

znode는 분산 애플리케이션을 구현할 때 유용한 몇 가지 속성을 가진다.

* 임시 znode
* 순차번호 (znode의 버전)
* 감시 (감시 기능 연산자와 트리거 연산자가 따로 있다 21.3.2에서 볼 수 있다.)

21.3.2 연산

또 다른 연산 중 multi도 있다. 여러개의 프리미티브 연산을 하나의 배치로 묶어서 전체 연산의 성공과 실패를 반환하는 방식이다.

이런 연산들은 자바와 C를 위한 바인딩으로 제공되고, contrib 바인딩에서 펄, 파이썬, REST 클라이언트도 지원한다. 동기, 비동기 API 모두 존재한다.

연산 중 exists, getChildren, getData에 감시를 설정할 수 있고, 쓰기 연산인 create, delete, setData에서 트리거된다. ACL은 감시의 대산이 아니다. 

znode는 ACL 목록과 함께 생성되는데, ACL은 권한을 결정한다. 주키퍼가 제공하는 인증체계는 digest, sasl, ip 이 있다.

21.3.3 구현

주키퍼 서비스는 두 가지 방식으로 운용할 수 있다. 독립 모드(고가용성이나 탄력성을 보장하지 않는 단일 주키퍼 서버)와 복제 모드 (앙상블이라는 컴퓨터 클러스터). 주키퍼는 고가용성을 복제를 통해 달성한다. 앙상블 내에서 과반수의 컴퓨터가 운영중인 동안에만 서비스를 제공한다. 

개념적으로 주키퍼는 매우 간단하다. 단지 znode 트리에 대한 모든 수정이 앙상블의 과반수 노드에 복제되도록 보장하는 것이다. 하지만 구현은 어렵다. 주키퍼는 두 단계로 동작하는 TCP에 의존하는 Zab라는 프로토콜을 사용하는데, 이것은 무한 반복될 수도 있기 때문이다.

단계1: 대표 선출 (Leader), 나머지 다른 서버는 추종자(follower). 과반수 또는 정족수 추종자의 상태가 대표와 동기화 되면 이 단계가 끝난다.
단계2: 원자적 브로드캐스트: 모든 쓰기 요청은 대표에게 forward, 대표는 추종자에 업데이트를 브로드 캐스트. 과반수 노드에서 변경을 저장하면 대표는 업데이트 연산을 커밋하고, 클라이언트는 성공 응답을 받는다.

앙상블의 모든 서버는 업데이트 사항을 자신의 znode 트리의 메모리 복사본에 기록하기 전에 디스크에 기록한다. 읽기는 모든 서버에 요청할 수 있고, 서버는 오로지 메모리에서 검색하기 때문에 속도가 매우 빠르다.

21.3.4 일관성

znode 트리에 수행된 모든 업데이트 연산은 zxid('Zookeeper transaction ID')라는 전체적으로 유일한 식별자를 부여받는다. 주키퍼에서 zxid인 z1이 z2보다 작으면 z1이 먼저 발생한것이다.

주키퍼 설계에서 보장하는 데이터 일관성 흐름에 대한 내용은 다음과 같다.
* 순차적 일관성: znode 업데이트 후에는 이전 값을 볼 수 없다
* 원자성: 업데이트는 성공 또는 실패다.
* 단일 시스템 이미지: 클라이언트는 연결된 서버에 관계없이 같은 시스템을 바라본다.
* 지속성: 업데이트 연산이 성공하면 업데이트 내역은 저장되고 취소되지 않는다.
* 적시성: 클라이언트가 시스템을 볼 때 지언이 발생해도 수십 초 이상 오래된 정보를 보여주지 않는다. 

성능상의 이유로 읽기는 주키퍼 서버의 메모리에서 이루어지고, 쓰기 순서를 전체 정렬하지는 않는다. 이러한 특성 때문에 주키퍼 밖의 메커니즘을 통해 통신하는 클라이언트가 일관되지 않은 주키퍼 상태를 보는 결과를 초래할 수 있다. 그러므로, 해당 클라이언트들은 sync를 호출해야한다.

21.3.5 세션

주키퍼 클라이언트는 앙상블의 서버 목록 설정을 가지고 있어야 한다. 시작 할 때 목록의 서버 중 하나로 연결을 시도하고, 실패하면 다른 서버에 시도한다. 모든 서버가 이용할 수 없는 상태면 결국 실패한다.

주키퍼 클라이언트가 서버에 연결되면 서버는 클라이언트를 위한 새로운 세션을 생성한다. 세션은 타임아웃 값을 가지고 있는데, 타임아웃은 세션을 생성한 애플리케이션이 결정한다. 세션이 타임아웃되어 만료되면 다시 열리지 않고, 세션에 관계된 모든 임시 znode도 사라진다. 클라이언트는 특정 시간 이상 세션이 일하지 않는 상태가 되면 그때마다 ping 요청을 보내 세션이 살아있도록 유지한다.(주키퍼 클라이언트 라이브러리가 자동으로 보낸다)

장애가 발생하여 다른 주키퍼 서버로 이동하는 것은 주키퍼 클라이언트 라이브러리가 자동으로 처리하는데, 끊어지고, 다시 접속되었음을 통지받을 수 있다.

주키퍼에는 시간과 관련된 몇 개의 인자가 있다. 틱타임은 주키퍼 시간의 기본 단위고, 앙상블 내의 서버는 수행하는 동작에 대한 스케줄링을 위해 틱타임을 사용한다. 일반적으로 2초다. 틱타임은 적절히 설정해야한다. 네트워크가 바쁠 때는 패킷이 늦게 도착하여 설정한 틱타임의 세션 타임아웃 시간보다 넘어가는 이슈가 생길 수 있기도 하다. 일반적으로 주키퍼 앙상블의 크기가 늘어날수록 세션 타임아웃도 늘어나야한다.

21.3.6 상태

ZooKeeper 객체는 생애 주기 내에서 다른 상태로 전이된다. 이 상태는 getState() 메서드를 호출하여 상태를 확인할 수 있다.

새로 생성된 ZooKeeper 인스턴스는 CONNECTING 상태, 연결이 성립되면 CONNECTED 상태가 된다. CONNECTING <-> CONNECTED 간에 상태 변화에 따라 Watcher로 통지를 받을 수 있다. (Watcher는 주키퍼 상태 변화와 znode의 변화 두 종류를 알기 위해 사용될 수 있다)

ZooKeeper 인스턴스는 close() 메서드가 호출되거나 세션이 타임아웃되어 KeeperState 값이 Expired로 변경되면 CLOSED로 전이되고, 다시 사용될 수 없다.