Ana içeriğe atla

Data Mesh Principles and Logical Architecture

 Data Mesh Principles and Logical Architecture The great divide of data What do we really mean by data? The answer depends on whom you ask. Today’s landscape is divided into  operational data  and  analytical data . Operational data sits in databases behind business capabilities served with microservices, has a transactional nature, keeps the current state and serves the needs of the applications running the business. Analytical data is a temporal and aggregated view of the facts of the business over time, often modeled to provide retrospective or future-perspective insights; it trains the ML models or feeds the analytical reports. The current state of technology, architecture and organization design is reflective of the divergence of these two data planes - two levels of existence, integrated yet separate. This divergence has led to a fragile architecture. Continuously failing ETL (Extract, Transform, Load) jobs and ever growing complexity of labyrinth of data pipel...
Yorum Gönder

Consistent Core

Consistent Core

Maintain a smaller cluster providing stronger consistency to allow large data cluster to coordinate server activities without implementing quorum based algorithms.



Problem

Linearizability is the strongest consistency guarantee where all the clients are guaranteed to see latest committed updates to data. Providing linearizability along with fault tolerance needs consensus algorithms like Raft, Zab or Paxos to be implemented on the servers.

While consensus algorithm is an essential requirement to implement a Consistent Core, there are various aspects of client interaction - such as how a client finds the leader, how duplicate requests are handled, etc - which are important implementation decisions. There are also some important implementation considerations regarding safety and liveness. Paxos defines only the consensus algorithm, but these other implementation aspects are not well documented in the Paxos literature. Raft very clearly documents various implementation aspects, along with a reference implementation and therefore is the most widely used algorithm today.

When a cluster needs to handle a lot of data, it needs more and more servers. For a cluster of servers, there are some common requirements, such as selecting a specific server to be the master for a particular task, managing group membership information, mapping of data partitions to the servers etc. These requirements need strong consistency guarantee, namely linearizability. The implementation also needs to be fault tolerant. A common approach is to use a fault-tolerant consensus algorithms based on Quorum. But in quorum-based systems throughput degrades with the size of the cluster.

Solution

Implement a smaller, 3 to 5 node cluster which provides linearizability guarantee as well as fault tolerance. [1] A separate data cluster can use the small consistent cluster to manage metadata and for taking cluster wide decisions with primitives like Lease. This way, the data cluster can grow to a large number of servers, but can still do certain actions which need strong consistency guarantees using the smaller metadata cluster.


Figure 1: Consistent Core

A typical interface of consistent core looks like this:

public interface ConsistentCore {

    CompletableFuture put(String key, String value);

 

    List<String> get(String keyPrefix);

 

    CompletableFuture registerLease(String name, long ttl);

 

    void refreshLease(String name);

 

    void watch(String name, Consumer<WatchEvent> watchCallback);

}

At the minimum, Consistent Core provides a simple key value storage mechanism. It is used to store metadata.

Metadata Storage

The storage is implemented using consensus algorithms such as Raft. It is an example of Replicated Write Ahead Log implementation, where replication is handled by Leader and Followers and High-Water Mark is used to track the successful replication using Quorum

Supporting hierarchical storage

Consistent Core is generally used to store data for things like: group membership or task distribution across servers. A common usage pattern is to scope the type of metadata with a prefix. e.g. for group membership, the keys will all be stored like /servers/1, servers/2 etc. For tasks assigned to servers the keys can be /tasks/task1, /tasks/task2. This data is generally read with all the keys with a specific prefix. For example, to get information about all the servers in the cluster, all the keys with prefix /servers are read.

An example usage is as following:

The servers can register themselves with the Consistent Core by creating their own key with prefix /servers.

client1.setValue("/servers/1", "{address:192.168.199.10, port:8000}");

 

client2.setValue("/servers/2", "{address:192.168.199.11, port:8000}");

 

client3.setValue("/servers/3", "{address:192.168.199.12, port:8000}");

The clients can then get to know about all the servers in the cluster by reading with key prefix /servers as following:

assertEquals(client1.getValue("/servers"), Arrays.asList("{address:192.168.199.12, port:8000}",

                                                            "{address:192.168.199.11, port:8000}",

                                                            "{address:192.168.199.10, port:8000}"));

Because of this hierarchical nature of data storage, products like [zookeeper], [chubby] provide a file system like interface, where users create directories and files, or nodes, with the concept of parent and child nodes. [etcd3] has a flat key space with the ability to get a range of keys.

Handling Client Interactions

One of the key requirements for Consistent Core functionality is how a client interacts with the core. The following aspects are critical for the clients to work with the Consistent Core.

Finding the leader

Serializability and Linearizability

When read requests are handled by follower servers, it is possible that clients can get stale data, as the latest commits from the leader have not reached the followers. The order in which the updates are received by the client is still maintained but the updates might not be most recent. This is the [serializability] guarantee as opposed to linearizability. Linearizability guarantees that every client gets the most recent updates. Clients can work with serializability guarantee when they just need to read metadata and can tolerate stale metadata for a while. For operations like Lease, linearizability is strictly needed.

If the leader is partitioned from the rest of the cluster, clients can get stale values from the leader, Raft describes a mechanism to provide linearizable reads. See for example etcd implementation of readIndex.

A similar situation can happen with followers which are partitioned. The follower may be partitioned and might not return the latest values to the client. To make sure that the followers are not partitioned and are up-to-date with the leader, they need to query the leader to know the latest updates, and wait till they receive the latest updates before responding to the client, See the proposed kafka design for example.

It's important that all the operations are executed on the leader, so a client library needs to find the leader server first. There are two approaches possible to fulfil this requirement.

§  The follower servers in the consistent core know about the current leader, so if the client connects to a follower, it can return the address of the leader. The client can then directly connect to the leader identified in the response. It should be noted that the servers might be in the middle of leader election when the client tries to connect. In that case, servers cannot return the leader address and the client needs to wait and try another server.

§  Servers can implement a forwarding mechanism and forward all the client requests to the leader. This allows clients to connect to any server. Again, if servers are in the middle of leader election, then clients need to retry until the leader election is successful and a legitimate leader is established.

Products like zookeeper and etcd implement this approach because they allow some read-only requests to be handled by the follower servers; this avoids a bottleneck on the leader when a large number of clients are read-only. This reduces complexity in the clients to connect to either leader or follower based on the type of the request.

A simple mechanism to find the leader is to try to connect to each server and try to send a request, the server responds with a redirect response if it's not the leader.

private void establishConnectionToLeader(List<InetAddressAndPort> servers) {

    for (InetAddressAndPort server : servers) {

        try {

            SingleSocketChannel socketChannel = new SingleSocketChannel(server, 10);

            logger.info("Trying to connect to " + server);

            RequestOrResponse response = sendConnectRequest(socketChannel);

            if (isRedirectResponse(response)) {

                redirectToLeader(response);

                break;

            } else if (isLookingForLeader(response)) {

                logger.info("Server is looking for leader. Trying next server");

                continue;

            } else { //we know the leader

                logger.info("Found leader. Establishing a new connection.");

                newPipelinedConnection(server);

                break;

            }

        } catch (IOException e) {

            logger.info("Unable to connect to " + server);

            //try next server

        }

    }

}

 

private boolean isLookingForLeader(RequestOrResponse requestOrResponse) {

    return requestOrResponse.getRequestId() == RequestId.LookingForLeader.getId();

}

 

private void redirectToLeader(RequestOrResponse response) {

    RedirectToLeaderResponse redirectResponse = deserialize(response);

    newPipelinedConnection(redirectResponse.leaderAddress);

 

    logger.info("Connected to the new leader "

            + redirectResponse.leaderServerId

            + " " + redirectResponse.leaderAddress

            + ". Checking connection");

}

 

 

private boolean isRedirectResponse(RequestOrResponse requestOrResponse) {

    return requestOrResponse.getRequestId() == RequestId.RedirectToLeader.getId();

}

Just establishing TCP connection is not enough, we need to know if the server can handle our requests. So clients send a special connection request for the server to acknowledge if it can serve the requests or else redirect to the leader server.

private RequestOrResponse sendConnectRequest(SingleSocketChannel socketChannel) throws IOException {

    RequestOrResponse request

            = new RequestOrResponse(RequestId.ConnectRequest.getId(), "CONNECT", 0);

    try {

        return socketChannel.blockingSend(request);

    } catch (IOException e) {

        resetConnectionToLeader();

        throw e;

    }

}

If an existing leader fails, the same technique is used to identify the newly elected leader from the cluster.

Once connected, the client maintains a Single Socket Channel to the leader server

Handling duplicate requests

In cases of failure, clients may try to connect to the new leader, resending the requests. But if those requests were already handled by the failed leader prior to failure, it might result in duplicates. Therefore, it's important to have a mechanism on the servers to ignore duplicate requests. Idempotent Receiver pattern is used to implement duplicate detection.

Coordinating tasks across a set of servers can be done by using Lease. The same can be used to implement group membership and failure detection mechanism.

State Watch is used to get notifications of changes to the metadata or time bound leases.

Examples

Google is known to use [chubby] lock service for coordination and metadata management.

[kafka] uses [zookeeper] to manage metadata and take decisions like leader election for cluster master. The proposed architecture change in Kafka will replace zookeeper with its own [raft] based controller cluster.

[bookkeeper] uses Zookeeper to manage cluster metadata.

[kubernetes] uses [etcd] for coordination, manage cluster metadata and group membership information.

All the big data storage and processing systems like [hdfs], [spark], [flink] use [zookeeper] for high availability and cluster coordination.

 

Yorumlar

Yorum Gönder

Bu blogdaki popüler yayınlar

Continuous Integration with Visual C++ and COM

  William E. Caputo ThoughtWorks Oren Miller ThoughtWorks July 2002 The Continuous Integration principles are applicable in many different development environments. While the principles remain largely unchanged from project to project, the implementation of this practice can vary considerably. Variables such as language, platform, team size & team location provide unique implementation challenges. Here we will outline how we've implemented CI in a COM/Windows environment for a project developing primarily in Visual C++. The More Often the Better What Is a Successful Build? Single Source Point Building the Code Self-Testing Code Automated Build Dependency Management What We Could Have Done Better Summing up The More Often the Better As noted in the main article, one of the least intuitive notions about integration is that less often does not result in less difficulty, it results in more difficulty. This is especially true when developing with C++. The build time on a development...
Yorum Gönder
Devamı

Rotation

  Rotation I've spent a lot of time of the last year wandering around ThoughtWorks, talking to lots of people on lots of projects. One message that's come home really firmly to me is the value of rotation. We practice rotation in lots of ways. One of the most notable is rotating around countries. We've put in a deliberate program to encourage people to spend 6-18 months in a different country. Living a good length of time in a different country does a huge amount to widen people's perspective of the world. I've benefitted personally from living both in the UK and USA, even though they are very similar cultures. This mental expansion is even greater for those that spend time in somewhere like India, where the cultural differences are greater. Geographic rotation presents lots of challanges, particular for older people with familes. One of the things we need to figure out is how to make geographic rotation easier for people, so more people do it. Already there's a...
Yorum Gönder
Devamı

Business Capability Centric

 Business Capability Centric A business-capability centric team is one whose work is aligned long-term to a certain area of the business. The team lives as long as the said business-capability is relevant to the business. This is in contrast to project teams that only last as long as it takes to deliver project scope. For example, an e-commerce business has capabilities such as buying and merchandising, catalog, marketing, order management, fulfilment and customer service. An insurance business has capabilities such as policy administration, claims administration, and new business. A telecom business has capabilities such as network management, service provisioning and assurance, billing, and revenue management. They may be further divided into fine-grained capabilities so that they can be owned by teams of manageable size. Business-capability centric teams are “think-it, build-it and run-it” teams. They do not hand over to other teams for testing, deploying or supporting what they...
Yorum Gönder
Devamı