Understanding Apache ZooKeeper Architecture and Installation with Hadoop Integration

Apache ZooKeeper is a critical component in the Hadoop ecosystem that enables building reliable, scalable distributed applications. As a centralized service for maintaining configuration information, naming, synchronization, and group services, ZooKeeper makes it easier to coordinate distributed processes and manage shared state across large clusters.

In this article, we‘ll dive deep into the architecture of ZooKeeper, understand how it integrates with Hadoop, and walk through the steps to install and configure ZooKeeper in a multi-node setup. We‘ll also explore common use cases and code examples to showcase ZooKeeper‘s capabilities.

What is Apache ZooKeeper?

Apache ZooKeeper is an open-source, distributed coordination service that enables synchronization and configuration maintenance across a cluster of nodes. It exposes a simple set of primitives that distributed applications can build upon to implement higher level services for synchronization, configuration maintenance, and naming.

ZooKeeper is designed to be highly reliable and scalable. It keeps all of its data in-memory and can scale to a large number of clients. The servers in the ZooKeeper ensemble (a cluster of ZooKeeper servers) communicate with each other to stay in sync and maintain a global view of the system state.

Why is ZooKeeper used in Hadoop?

In a distributed system like Hadoop that runs on large clusters with many nodes, coordination and synchronization between various processes and services becomes critical. Failures are common at scale and the system needs built-in mechanisms to maintain consistency and availability.

This is where ZooKeeper comes into the picture. Hadoop leverages ZooKeeper for:

• Coordination between NameNode and DataNodes in HDFS
• ResourceManager HA (High Availability) in YARN
• Coordination and configuration management in HBase
• Implementing service discovery and leader election
• Storing and managing configuration for various services
• Providing distributed synchronization primitives like locks and barriers

By externalizing these critical functions to ZooKeeper, Hadoop projects can focus on their core functionalities while ZooKeeper takes care of coordination and synchronization in a reliable manner. ZooKeeper makes Hadoop clusters more resilient, manageable, and scalable.

ZooKeeper Architecture Overview

Let‘s now look under the hood and understand ZooKeeper‘s architecture. At a high level, ZooKeeper comprises of two main components:

• ZooKeeper servers (called Quorum Peers) that maintain the state
• ZooKeeper clients that connect to the servers

ZooKeeper Quorum Peers and Znodes

ZooKeeper has a hierarchical namespace (similar to a filesystem) called the data tree where each node is referred to as a znode. These znodes store the configuration and state information.

Znodes are the core abstraction in ZooKeeper. They have a few important characteristics:

• Znodes can store data (limited to 1MB by default)
• Znodes maintain a stat structure with version information and timestamps
• Znodes can be ephemeral (they disappear when client disconnects) or persistent
• Znodes can have child znodes, making the structure hierarchical like a filesystem

The servers (Quorum Peers) that make up the ZooKeeper ensemble aim to provide a replicated and consistent view of the znodes and overall system state. Write requests (to create/modify znodes) go through a single server which acts as a leader, while reads can be served by any of the servers. Servers communicate with each other to stay in sync.

ZooKeeper Client

Clients connect to a single ZooKeeper server at a time. They maintain a TCP connection with the server, through which they send requests and receive responses. If a client loses connectivity to a server, it can seamlessly connect to a different server.

For read requests, the client can connect to any of the ZooKeeper servers. Since the ZooKeeper data is replicated and kept in-sync across all servers, any server can service the read.

However, for write requests, all requests flow through the leader. When the leader receives a write request, it first comes to an agreement with the other servers using a protocol called Zab (ZooKeeper Atomic Broadcast) before committing the update. This provides strong consistency guarantees.

Watchers and Notifications

One of the powerful features of ZooKeeper is the ability to set Watches on znodes. Clients can set a watch while reading data, and get notified when the data of that znode changes. This allows clients to react to changes without having to poll the servers.

Installing and Configuring ZooKeeper

Now that we have a fair understanding of ZooKeeper architecture, let‘s go through the steps to install and configure ZooKeeper.

Prerequisites

• Java 8 or above installed

Steps

1. Download ZooKeeper distribution from the official Apache ZooKeeper downloads page (https://zookeeper.apache.org/releases.html). As of March 2023, the latest stable version is 3.8.1.

2. Extract the downloaded tarball:

   tar -xvf apache-zookeeper-3.8.1-bin.tar.gz

3. Rename the extracted directory for easier reference:

   mv apache-zookeeper-3.8.1-bin zookeeper

4. Create a data directory to store ZooKeeper data:

   mkdir zookeeper/data

5. Navigate to the conf directory and create a file named zoo.cfg:

   cd zookeeper/conf
   cp zoo_sample.cfg zoo.cfg

6. Open zoo.cfg in a text editor and set the following minimum required properties:

   tickTime=2000
   dataDir=/path/to/zookeeper/data
   clientPort=2181

   Here, tickTime is the basic time unit in milliseconds, dataDir is the location to store in-memory database snapshots, and clientPort is the port to listen for client connections.

7. For a multi-server setup, add the following lines to the configuration file:

   server.1=<server1_hostname>:2888:3888
   server.2=<server2_hostname>:2888:3888
   server.3=<server3_hostname>:2888:3888

   Here, 1, 2 and 3 are unique identifiers for each server in the ensemble. The hostname is where each server is running, 2888 is the port for followers to connect to the leader, and 3888 is the port for leader election.

8. Create a file named "myid" in the data directory of each server, and put the server number (1, 2, or 3 in this case) in it.

9. Start the ZooKeeper servers:

   bin/zkServer.sh start

10. Connect to ZooKeeper using the CLI:

     bin/zkCli.sh -server 127.0.0.1:2181

That‘s it! With these steps, you will have a basic ZooKeeper setup ready. For a production deployment, there are additional configuration options to consider like security, performance tuning, and monitoring.

ZooKeeper Use Cases and Examples

Let‘s now look at some common use cases of ZooKeeper and code snippets to interact with ZooKeeper.

Leader Election

One of the common use cases of ZooKeeper is to implement leader election in a distributed system. The idea is to have a single coordinator (leader) at any given time. Here‘s a simplified example in Java:

// Create a leader election znode
zk.create("/election", data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

// Get children of election znode
List<String> children = zk.getChildren("/election", false);

// Sort children to determine the leader 
Collections.sort(children);
String leader = children.get(0);

// Watch for changes to the election znode
zk.getChildren("/election", event -> {
    if (event.getType() == Watcher.Event.EventType.NodeChildrenChanged) {
        // Re-elect leader if needed
    }
});

Here, each client creates an ephemeral sequential znode under the "/election" znode. The client with the lowest sequence number becomes the leader. Clients watch the "/election" znode for changes, and if the leader goes away (ephemeral node disappears), a new leader is elected.

Locks and Synchronization

ZooKeeper can also be used to implement distributed locks and synchronize access to shared resources. Here‘s an example of a simple lock implementation:

// Create a lock znode
zk.create("/lock", data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

// Get children of lock znode
List<String> children = zk.getChildren("/lock", false);

// Sort children 
Collections.sort(children);

// Get the znode name of the client holding the lock
String lockHolder = children.get(0);

// Watch the znode of the client preceding the current client
String precedingClient = "/lock/" + children.get(children.indexOf(currentClient) - 1);
zk.exists(precedingClient, event -> {
    if (event.getType() == Watcher.Event.EventType.NodeDeleted) {
        // Acquire lock if the preceding client releases it
    }
});

The idea here is similar to leader election. Each client creates an ephemeral sequential znode under "/lock". The client with the lowest sequence number holds the lock. Other clients watch the znode of the client preceding them, and when that znode disappears (indicating the lock is released), they attempt to acquire the lock.

Message Queue

ZooKeeper can also be used as a simple message queue. Producers can create sequential znodes with the message data, and consumers can process the messages in order. Here‘s a simplified producer-consumer example:

// Producer
zk.create("/queue/message", message, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);

// Consumer
while (true) {
    List<String> children = zk.getChildren("/queue", false);
    if (children.isEmpty()) {
        // Wait for messages
        continue;
    }

    // Get the oldest message
    String oldest = children.get(0);
    byte[] data = zk.getData("/queue/" + oldest, false, null);

    // Process the message
    processMessage(data);

    // Remove the processed message
    zk.delete("/queue/" + oldest, -1);
}

The producer creates a persistent sequential znode for each message. The consumer gets the list of znodes, processes the oldest message (lowest sequence number), and deletes the corresponding znode.

These are just a few examples to give you a flavor of what‘s possible with ZooKeeper. In practice, ZooKeeper is used for a wide variety of coordination and synchronization tasks in distributed systems.

Conclusion

In this article, we took a deep dive into Apache ZooKeeper, a critical component in the Hadoop ecosystem and distributed systems in general. We looked at the architecture of ZooKeeper, how it integrates with Hadoop, and went through the steps to install and configure ZooKeeper.

We also explored some common use cases of ZooKeeper like leader election, locks, and message queues, along with code examples.

ZooKeeper provides a simple yet powerful set of primitives for coordination and synchronization in distributed systems. By externalizing these critical functions to ZooKeeper, distributed applications can focus on their core functionalities while ZooKeeper takes care of coordination in a reliable and scalable manner.

While we covered the basics here, ZooKeeper has many more features and configuration options. It‘s a versatile tool that can be used in a variety of ways to build reliable distributed systems.

I hope this article gave you a good understanding of ZooKeeper and how it fits into the Hadoop ecosystem. Happy distributing!