Designing data-intensive Applications Chapter 1: Reliability, Scalability, and Maintainability

The Internet has done so well that most people see it as a natural resource, like the ocean, rather than an artifact. When was the last time you had this kind of large-scale, error-free technology?

– Alan Kay in An interview with Dr. Dobb magazine (2012)

[TOC]

Many applications today are data-intensive rather than compute-intensive. As a result, CPUS are rarely the bottleneck for such applications, and the larger problem is usually the volume of data, the complexity of the data, and the speed at which the data changes.

Data-intensive applications are often built from standard components that provide many common functions; For example, many applications require:

Database

Store data so that you or other applications can find it again later

Cache

Remember the results of expensive operations to speed up reads

Search indexes

Allows users to search for data by keyword or filter data in a variety of ways

Stream processing

Send messages to other processes for asynchronous processing

Batch processing

Periodically compress large amounts of accumulated data

If these features sound mundane, it’s because these data systems ** are very successful abstractions that we use without thinking and take for granted. Most engineers don’t want to write a storage engine from scratch, because a database is a perfect tool for developing applications.

But it’s not that simple. Different applications have different needs, so the database system is also a hundred flowers bloom, with a variety of characteristics. There are a number of ways to do caching, a number of ways to do search indexes, and so on. Therefore, before developing an application, it is important to understand the tools and methods that are best suited to the task at hand. And when a single tool doesn’t solve your problem, you’ll find it difficult to use them in combination.

This book will be a journey through the principles, practices, and applications of data systems, as well as a methodology for designing data-intensive applications. We will explore the commonalities and features of different tools and how they are implemented.

This chapter will start with the basic goal we want to achieve: reliable, scalable, and maintainable data systems. We will clarify what these terms mean and outline ways to consider these goals. And review some of the basics required for subsequent chapters. In the sections that follow, we’ll delve into the design decisions that can be made when designing data-intensive applications.

Thinking about data systems

We generally think of tools like databases, message queues, caches, and so on as falling into several distinct categories. Although databases and message queues have some superficial similarities — they both store data for a period of time — they have very different access patterns, meaning very different performance characteristics and implementations.

So why are we lumping these things together under the umbrella of data systems?

In recent years, there are many new data storage tools and data processing tools. They are optimized for different application scenarios and therefore no longer fit neatly into traditional categories [1]. The lines between categories are becoming blurred, for example: data stores can be used as message queues (Redis), and message queues have database-like persistence guarantees (Apache Kafka).

Second, more and more applications have a wide range of stringent requirements, and a single tool is not sufficient to meet all data processing and storage requirements. Instead, the overall effort is broken down into a series of tasks that can be efficiently accomplished by a single tool and stitched together by application code.

For example, if caching (an application-managed caching layer, Memcached or its equivalent) and full-text search (a full-text search server, such as Elasticsearch or Solr) are separated from the primary database, it is usually the application code’s responsibility to keep the cache/index in sync with the primary database. Figure 1-1 shows what this architecture might look like (details will be covered in a later section).

reliability

People have an intuitive idea of whether something is reliable or not. Typical expectations for reliable software include:

• The application performs what the user expects.
• Allow users to make mistakes and use the software in unexpected ways.
• Under the expected load and data volume, the performance meets the requirements.
• The system protects against unauthorized access and abuse.

If all this together means “working correctly,” then reliability can be loosely interpreted as “continuing to work correctly even when things go wrong.”

The cause of an error is called a fault. A system that can anticipate and respond to a fault can be fault-tolerant or resilient. The term “fault tolerance” can be misleading because it implies that the system can tolerate all possible errors, but in practice this is impossible. For example, if the entire earth (and every server on it) were swallowed by a black hole, the ability to tolerate such errors would require hosting the web in space — good luck getting that budget approved. So when you’re talking about fault tolerance, it only makes sense to talk about specific types of errors.

Note that a fault is not the same as a failure [2]. Failure is usually defined as a part of the system that deviates from its standard, whereas failure is when the system as a whole stops providing services to users. The probability of failure cannot be reduced to zero, so it is best to design a fault tolerant mechanism to prevent failure caused by failure. This book will introduce several techniques for building reliable systems from unreliable components.

Counterintuitively, in such fault-tolerant systems, it makes sense to increase failure rates by deliberately triggering, for example, killing individual processes at random without warning. Many high-risk vulnerabilities are actually caused by poor error handling [3], so we can deliberately cause failures to ensure that fault tolerance mechanisms are constantly running and tested, thus increasing confidence that the system can handle failures correctly when they occur naturally. Netflix’s Chaos Monkey [4] is an example of this approach.

Although we are generally more inclined to tolerate errors than prevent errors. But there are cases where prevention is better than cure (for example, when no cure exists). Such is the case with security. For example, if an attacker has compromised the system and gained access to sensitive data, it cannot be undone. But this book focuses on the kinds of failures that can be recovered, as described in the following sections.

A hardware failure

Software error

We often think of hardware failures as random and independent: just because a disk fails on one machine doesn’t mean a disk on another machine will fail. A large number of hardware components cannot fail at the same time unless they are weakly correlated (the same reason that causes the correlation error, such as the temperature of the server rack).

Another type of error is internal systematic error [7]. Such errors are unpredictable and, because they are cross-node related, tend to cause more system failures than unrelated hardware failures [5]. Examples include:

• A BUG that causes all application server instances to crash when certain incorrect input is accepted. The leap second on June 30, 2012, for example, crashed many applications at the same time due to a bug in the Linux kernel.
• Runaway processes consume shared resources, including CPU time, memory, disk space, or network bandwidth.
• The services on which the system depends become slow, unresponsive, or start returning incorrect responses.
• Cascading faults, in which a small fault in one component triggers a fault in another component, which triggers more failures [10].

The bugs that cause such software failures often lie dormant for a long time until they are triggered by an exception. This means that the software makes some assumption about its environment that is generally true, but for some reason no longer holds [11].

While there is no quick fix for systemic failures in software, there are a number of small fixes that can be taken, such as: carefully considering the assumptions and interactions in the system; A thorough test; Process isolation; Allow processes to crash and restart. Measure, monitor and analyze system behavior in production environment. If the system can provide some assurance (for example, in a message queue, the number of incoming messages equals the number of outgoing messages), the system can constantly check itself at run time and raise an alarm in case of discrepancies (In Favor) ** [12].

Human error

How important is reliability?

Reliability is not just for nuclear power plants and air traffic control software; we expect it to work reliably for many mundane applications. Errors in business applications can lead to lost productivity (and perhaps legal risks with incomplete data reporting), while disruptions to e-commerce sites can lead to significant losses in revenue and reputation.

Even in “non-critical” applications, we have a responsibility to our users. Imagine a parent storing all their photos and videos of their children in your photos app [15]. How would they feel if the database suddenly became corrupted? Are they likely to know how to recover from a backup?

In some cases, we may choose to sacrifice reliability to reduce development costs (such as prototyping products for an unproven market) or operating costs (such as extremely low-margin services), but we should be aware of what we are doing when we cut corners.

scalability

Describe the load

Discuss growth (what happens if the load doubles?) Be able to briefly describe the current load on the system. Loads can be described by numbers called load parameters. Depending on the system architecture, the best choice of parameters could be requests to the Web server per second, read/write ratios in the database, number of simultaneous active users in the chat room, cache hit ratio, or something else. Beyond that, maybe averages are important to you, or maybe your bottleneck is a few extreme scenarios.

To make this concept more concrete, we take the data released by Twitter in November 2012 [16] as an example. Twitter’s two main businesses are:

Release tweets

Users can post new messages to their followers (average 4.6K requests/SEC, peak over 12K requests/SEC).

Homepage Timeline

Users can look up tweets from people they follow (300K requests/second).

Handling 12,000 writes per second (peak tweet rate) is straightforward. Twitter’s scalability challenge, however, comes not primarily from the volume of tweets, but from fan-outs — where each user follows many people and is followed by many people.

[^ii]: Fan out: term borrowed from electrical engineering, describing the number of logical gates in which the input is connected to the output of another gate. The output needs to provide enough current to drive all connected inputs. In transactional systems, we use it to describe the number of requests that need to be performed for other services in order to service one incoming request.

In general, this pair of operations can be implemented in two ways.

1. To post a tweet, simply insert the new tweet into the global tweet collection. When a user requests a timeline for their homepage, they first look up all the people they follow, query the tweets from those users and merge them chronologically. In a relational database, as shown in Figure 1-2, you can write queries like this:

   SELECT tweets.*, users.* 
     FROM tweets 
     JOIN users   ON tweets.sender_id = users.id 
     JOIN follows ON follows.followee_id = users.id 
     WHERE follows.follower_id = current_userCopy the code

   

   Figure 1-2 Simple implementation of the relational mode of twitter home page timeline

2. Maintain a cache for each user’s home page timeline, like each user’s Tweet inbox (Figure 1-3). When a user posts a tweet, look up all the people who follow that user and insert the new tweet into each homepage timeline cache. So the request to read the home page timeline is minimal because the results are already calculated in advance.

   

   Figure 1-3 Data pipeline for distribution of Tweets to followers, Load parameters in November 2012 [16]

Describe the performance

Once the load on the system can be described, what happens when the load increases can be investigated. There are two ways to look at it:

• What happens to system performance when you increase load parameters and keep system resources (CPU, memory, network bandwidth, and so on) constant?
• How many system resources do YOU need to add when you add load parameters and want to keep performance constant?

Both of these issues require performance data, so let’s briefly look at how to describe system performance.

For batch systems like Hadoop, the usual concerns are throughput, the number of records that can be processed per second, or the total time it takes to run a job on a data set of a particular size [^ III]. For online systems, what is usually more important is the response time of the service, which is the time between the client sending the request and receiving the response.

[^ III]: Ideally, the running time of a batch job is the size of the data set divided by the throughput. In practice, running times tend to be longer due to data skew (data is not evenly distributed across each worker process), which requires waiting for the slowest task to complete.

Latency and response time

Latency and response time are often used synonymously, but they are not the same. Response time is what the customer sees, including network latency and queuing latency in addition to the time it takes to actually process the request (service time). Latency is the duration of a request waiting to be processed, during which it is in dormant and waiting for service [17].

Percentile sites in practice

High percentiles become particularly important in multi-invoked back-end services. Even if invoked in parallel, the end user request still needs to wait for the slowest parallel call to complete. As shown in Figure 1-5, it only takes one slow call to slow down the entire end user request. Even if only a small percentage of back-end calls are slower, the chance of getting slower calls increases if end user requests require multiple back-end calls, so a higher percentage of end user requests are slower (an effect known as tail-delay amplification [24]).

If you want to add response time percentage points to your service’s monitoring dashboard, you need to continuously calculate them efficiently. For example, you might want to keep a scrolling window of request response times within the last 10 minutes. Every minute, you calculate the median and various percentages in the window and plot those measurements on a graph.

The simple implementation is to keep a list of response times for all requests in a time window and sort the list every minute. If efficiency is too low for you, there are algorithms that can approximate percentiles with minimal CPU and memory costs (such as forward attenuation [25], T-Digest [26], or HdrHistogram [27]). Note that averaging percentages (e.g., reducing time resolution or merging data from multiple machines) does not make mathematical sense – the correct way to aggregate response time data is to add histograms [28].

Ways to deal with loads

We have now discussed the parameters used to describe load and the metrics used to measure performance. It’s time to start talking seriously about scalability: How do you maintain good performance as load parameters increase?

An architecture that accommodates a certain level of load is unlikely to be able to handle 10 times that. If you are developing a rapidly growing service, you may need to rethink your architecture every time the load increases by an order of magnitude — or more often.

There is a lot of talk about the opposition between vertical scaling up (moving to more powerful machines) and horizontal scaling out (spreading the load across many small machines). Distributing load across multiple machines is also known as a “shared-nothing” architecture. Systems that can run on a single machine are generally simpler, but high-end machines can be very expensive, so very dense loads often inevitably require lateral scaling. Good architecture in the real world requires a pragmatic combination of these two approaches, as it can be simpler and cheaper to use a few sufficiently powerful machines than to use a large number of small virtual machines.

Some systems are elastic, which means they can automatically increase computing resources when they detect an increase in load, while others scale manually (manually analyzing capacity and deciding to add more machines to the system). An elastic system can be useful if the load is highly unpredictable, but scaling up the system manually is simpler and may result in fewer unexpected operations (see “Rebalancing partitions”).

Deploying stateless services ** across multiple machines is simple, but moving a stateful data system from a single node to a distributed configuration can introduce a lot of additional complexity. For this reason, common sense dictates that the database should be placed on a single node (vertically scaled) until scaling costs or availability requirements force it to be distributed.

As the tools and abstractions of distributed systems get better, this common sense may change, at least for some types of applications. You can expect distributed data systems to become the default in the future, even for scenarios that don’t handle large amounts of data or traffic. The rest of the book introduces a variety of distributed data systems, discussing not only their scalability, but also their ease of use and maintainability.

Large-scale systems architectures are often application-specific — no one-size-fits-all extensible architecture (informally: Magic Scaling sauce). The problem with an application can be the amount of reads, writes, data to store, complexity of data, response time requirements, access patterns, or a hodgepodge of all.

For example, a system handling 100,000 requests per second (each 1 kB in size) will look very different from a system handling three requests per minute (each 2GB in size), even though both systems have the same data throughput.

An extensible architecture that works well for applications is built around the assumption that ** : Which operations are common? Which operations are rare? This is called a load parameter. If the assumptions turn out to be wrong, then the engineering investment made for the expansion is wasted, and at worst, counterproductive. In early stage startups or informal products, the ability to support rapid iterations of the product is often more important than a hypothetical load that can scale into the future.

Although these architectures are application-specific, extensible architectures are often built from common building blocks and arranged in common patterns. In this book, we will discuss these artifacts and patterns.

maintainability

Operability: Life is short, care for operation and maintenance

According to one, “Good operations can often bypass the limitations of junk (or incomplete) software, and even good software cannot run reliably with junk”. Although some aspects of operations can and should be automated, it is still up to people to set up the automation mechanisms that get it right in the first place.

Operations teams are critical to keeping software systems running smoothly. Typical responsibilities of a good operations team are as follows (or more) [29] :

• Monitor the health of the system and quickly restore service in case of poor service status
• Trace the cause of the problem, such as system failure or performance degradation
• Keep your software and platform up-to-date, such as security patches
• Understand the interactions between systems so that you can circumvent abnormal changes before they cause damage.
• Anticipate future problems and solve them before they arise (e.g., capacity planning)
• Establish deployment, configuration, and management good practices and write tools
• Perform complex maintenance tasks, such as migrating applications from one platform to another
• Maintains system security when configuration changes
• Define workflows to make operations predictable and maintain a stable production environment.
• Iron camp flow of soldiers, maintain the organization to understand the system.

Good operability means easier day-to-day work so that operations teams can focus on high-value things. Data systems can make everyday tasks easier in a variety of ways:

• Provide visibility into the internal state of the system and runtime behavior through good monitoring
• Provides good support for automation and integration of systems with standardized tools
• Avoid reliance on a single machine (allow machine downtime for maintenance while the entire system continues to operate uninterrupted)
• Provide good documentation and an easy-to-understand action model (” If you do X, Y happens “)
• Provides good default behavior, but also allows administrators the freedom to override defaults if needed
• Self-repair when available, but also allow administrators to manually control system state when needed
• Predictable behavior minimizes accidents

Simplicity: Manage complexity

Evolvability: Embrace change

The requirements of the system are always the same and almost impossible. More likely, they are in a state of constant change, for example: you learn new facts, unexpected application scenarios emerge, business priorities change, users demand new features, new platforms replace old ones, legal or regulatory requirements change, system growth forces architectural changes, etc.

In terms of organizational processes, Agile work patterns provide a framework for adapting to change. The Agile community has also developed technical tools and patterns, such as Test-driven development (TDD) and Refactoring, that are useful for developing software in a frequently changing environment.

Most of the discussion of these agile techniques focuses on a fairly small scale (a few code files in the same application). This book will explore ways to improve agility at the level of larger data systems, which may consist of several different applications or services. For example, how would you “refactor” the architecture of Twitter in order to change the method of assembling the homepage timeline from method 1 to method 2?

The ease with which a data system can be modified and adapted to changing requirements is closely related to simplicity and abstraction: simple and understandable systems are often easier to modify than complex ones. But because this is such an important concept, we will use a different term to refer to agility at the data system level: evolvability [34].

The summary of this chapter

reference

1. Michael Stonebraker and Uğur Cetintemel: ‘One Size Fits All’: An Idea Whose Time Has Come and Gone, “at 21st International Conference on Data Engineering (ICDE), April 2005.

2. Walter L. Heimerdinger and Charles B. Weinstock: “A Conceptual Framework for System Fault Tolerance,” Technical Report CMU/SEI-92-TR-033, Software Engineering Institute, Carnegie Mellon University, October 1992.

3. Ding Yuan, Yu Luo, Xin Zhuang, et al.: “Simple Testing Can Prevent Most Critical Failures: An Analysis of Production Failures in Distributed data-intensive Systems,” at 11th USENIX Symposium on Operating Systems Design and Implementation (OSDI), October 2014.

4. Yury Izrailevsky and Ariel Tseitlin: “The Netflix Simian Army,” techblog.netflix.com, July 19, 2011.

5. Daniel Ford, François Labelle, Florentina I. Popovici, et al.: “Availability in Globally Distributed Storage Systems,” at 9th USENIX Symposium on Operating Systems Design and Implementation (OSDI), October 2010.

6. Brian Beach: “Hard Drive Reliability Update — Sep 2014,” Backblaze.com, September 23, 2014.

7. Laurie Voss: “AWS: The Good, the Bad and the Ugly,” blog.awe.sm, December 18, 2012.

8. Haryadi S. Gunawi, Mingzhe Hao, Tanakorn Leesatapornwongsa, et al.: “What Bugs Live in the Cloud? At 5 th, “ACM Symposium on Cloud Computing (SoCC), November 2014. The doi: 10.1145/2670979.2670986

9. Nelson Minar: “Leap Second Crashes Half the Internet,” somebits.com, July 3, 2012.

10. Amazon Web Services: “Summary of the Amazon EC2 and Amazon RDS Service Disruption in the US East Region,” aws.amazon.com, April 29, 2011.

11. Richard I. Cook: “How Complex Systems Fail,” Cognitive Technologies Laboratory, April 2000.

12. Jay Kreps: “Getting Real About Distributed System Reliability,” empathybox.com, March 19, 2012.

13. David Oppenheimer, Archana Ganapathi, and David A. Patterson: Why Do Internet Services Fail, and What Can Be Done About It? , “at the 4th USENIX Symposium on Internet Technologies and Systems (USITS), March 2003.

14. Nathan Marz: “Principles of Software Engineering, Part 1,” nathanmarz.com, April 2, 2013.

15. Michael Jurewitz: “The Human Impact of Bugs,” Jury. Me, March 15, 2013.

16. Raffi Krikorian: “Timelines at Scale,” at QCon San Francisco, November 2012.

17. Martin Fowler: Patterns of Enterprise Application Architecture. Addison Wesley, 2002. ISBN: 978-0-321-12742-6

18. Kelly Sommers: “After all that run around, what caused 500ms disk latency even when we replaced physical server?” twitter.com, November 13, 2014.

19. Giuseppe DeCandia, Deniz Hastorun, Madan Jampani, et al.: “Dynamo: Amazon’s Highly Available Key-Value Store,” at 21st ACM Symposium on Operating Systems Principles (SOSP), October 2007.

20. Greg Linden: “Make Data Useful,” slides from presentation at Stanford University Data Mining class (CS345), December 2006.

21. Tammy Everts: “The Real Cost of Slow Time vs Downtime,” webperformancetoday.com, November 12, 2014.

22. Jake Brutlag: “Speed Matters for Google Web Search,” googleresearch. Blogspot. Co. UK, June 22, 2009.

23. Tyler Treat: “Everything You Know About Latency Is Wrong,” bravenewgeek.com, December 12, 2015.

24. Jeffrey Dean and Luiz Andre Barroso: “The Tail at Scale,” Communications of the ACM, volume 56, number 2, pages 74-80, February 2013. The doi: 10.1145/2408776.2408794

25. Graham Cormode, Vladislav Shkapenyuk, Divesh Srivastava, and Bojian Xu: “Forward Decay: A Practical Time Decay Model for Streaming Systems,” at 25th IEEE International Conference on Data Engineering (ICDE), March 2009.

26. Ted Dunning and Otmar Ertl: “Computing Extremely Accurate Quantiles Using t-Digests,” github.com, March 2014.

27. Gil Tene: “HdrHistogram,” hdrhistogram.org.

28. Baron Schwartz: “Why Percentiles Don’t Work the Way You Think,” vividcortex.com, December 7, 2015.

29. James Hamilton: “On Designing and Deploying Internet-Scale Services,” at 21st Large Installation System Administration Conference (LISA), November 2007.

30. Brian Foote and Joseph Yoder: “Big Ball of Mud,” at the 4th Conference on Pattern Languages of Programs (PLoP), September 1997.

31. Frederick P Brooks: “No Silver Bullet — Essence and Accident in Software Engineering,” in The Mythical Man-Month, Anniversary edition, Addison-Wesley, 1995. ISBN: 978-0-201-83595-3

32. Ben Moseley and Peter Marks: “Out of the Tar Pit,” at BCS Software Practice Advancement (SPA), 2006.

33. Rich Hickey: “Simple Made Easy,” at Strange Loop, September 2011.

34. Hongyu Pei Breivold, Ivica Crnkovic, and Peter J. Eriksson: “Analyzing Software Evolvability,” at 32nd Annual IEEE International Computer Software and Applications Conference (COMPSAC), Out 2008. Doi: 10.1109 / COMPSAC. 2008.50