CT Wu

Does it matter?

This week, in our study group, there is an interesting question.

If our service use MongoDB as the primary database and leverage ElasticSearch to benefit searching. Each request is in, our service is responsible for a dual write to update both MongoDB and ElasticSearch. Is this implementation considered CQRS?

In fact, this implementation is indeed generating a read model for searching and a write model for domain operations. If there is any search request, we can build a search layer or a search service for that.

This looks like CQRS, doesn’t it?

In my opinion, such an implementation is not CQRS.

There are two core concepts about CQRS.

1. event sourcing
2. responsibility segregation

Let’s see their actual meaning.

Event sourcing

Event sourcing is the most important concept for CQRS. When we have a new feature requirement, we can leverage the existed events and replay to generate a new read model so that we don’t need to modify the original code base.

In other words, we can easily fulfill any query requirement without impact the production functionality.

Let’s go back to the original example.

If our service keeps every input event in MongoDB so that new read models can be generated in the future, then I would say it’s “very much like” CQRS.

Responsibility segregation

The another key point is the responsibility segregation, i.e., the last two letters of CQRS.

What does that mean?

From my point of view, responsibility for building the read and write models must be separated.

If our service is responsible for both update MongoDB and update ElasticSearch, that doesn’t sound separate at all.

In order to release the burden of updating many models, CQRS introduces an event synchronization mechanism for event consumers to generate read models and make them consistent eventually.

Let’s continue with the original example.

If our service maintain the MongoDB’s domain states and keep every domain operation as audit logs. Then, we can build a worker who regularly digest these logs and update ElasticSearch. So, this is a CQRS definitely.

Conclusion

Why do we need to clarify these terminology details? Is it important to know if it is considered CQRS?

Well, it depends.

If we want to make the original system better, then we need to know exactly what a paradigm means. If we hastily assume that the original system is already a paradigm, then we miss the opportunity to look at it deeply.

The original implementation actually works well at the moment. However, if the input request increases, updating ElasticSearch will become a bottleneck while handling domain operations. At that time, we then have to make additional improvements.

Those paradigms are the indexes that provide how to improve. Because of this, we need to understand the specifics of the paradigms.

Originally published on Medium

Explaining how to achieve end-to-end exactly-once guarantee

Previously, when discussing message queues, we mentioned that when selecting a message queue, we should consider its delivery guarantee. Depending on the needs, there are different requirements for the delivery guarantee. The more critical the application, the higher level of guarantee is required.

There are three different levels of delivery guarantees, ranging from weak to strong as follows.

1. At most once
2. At least once
3. Exactly once

When we define a delivery guarantee, in fact, we need to understand the scope of the guarantee. In an end-to-end delivery guarantee, there are three different paths that work together to define a complete end-to-end delivery guarantee.

These three paths are as follows.

• Producer Perspective
• Consumer Perspective
• Sink Perspective

An end-to-end delivery guarantee is actually determined by the weakest of these three paths. For example, if the producer has an exactly-once guarantee, but the consumer can only achieve at-least once, and the sink has at-least once too, then end-to-end is at-least once guaranteed.

In this article, we will focus on how to achieve an exactly-once guarantee. Therefore, it is necessary to understand the meaning of these three paths and how to handle failures.

Producer Perspective

From the producer’s point of view, we always want any message that is sent to be delivered.

Most message queues are able to tell the producer whether the message was sent successfully or not, and when the message is sent successfully, the producer will receive a response. We can be sure that if we receive a response, it was successful, and if we don’t receive it, it’s basically a failure.

However, if we just lose the response but send it successfully, we have a problem.

The diagram above shows a common error process, so how do we avoid sending twice?

There are several approaches.

The first approach is for the producer to maintain a unique constraint on the message, e.g. (producer_id, msg_id). When a send failure occurs, the producer goes to the broker and queries the broker’s internal storage to determine whether the send failed or the response was lost.

Nevertheless, not every message queue supports this approach. If the message disappears as soon as it is taken away by the consumer, such as RabbitMQ, then this approach will not work. Conversely, systems like Kafka, which store messages as logs, can provide producers with queries.

In the second approach, the producer does not identify if a response is lost, instead resending the message anyway, but leaving a GUID (globally unique ID) on the message. On the other hand, the consumer receives the message and has to determine if the GUID has been processed, or simply ignore it if it has been processed.

If using Kafka, there is another approach to guarantee exactly-once, that is, through message transactions.

Just like a relational database, declare START TRANSACTION before sending a message, and declare COMMIT after the message is sent. If any of the intermediate messages fail, then the producer does not COMMIT but simply aborts the entire transaction and resends it. In this way, a failed resend does not result in a duplicate send.

Consumer Perspective

This is the scenario that all message queues focus on. The delivery guarantee provided by the message queue is only a part of the overall end-to-end delivery guarantee from the consumer perspective.

Most message queues can only provide at-least once guarantee. Once the consumer has correctly processed the message, an acknowledgment is sent to the message queue, and the message is removed only when the message queue receives the acknowledgment.

And, the flow of the problem is as follows.

When Consumer 1 handles Message A and an acknowledgment is lost, then Consumer 2 can continue to get the same Message A and handle it, resulting in handling twice.

Basically, this is a common problem for messages queues that are supported at-least once.

Idempotent Updates

In general, developers try to make messages as idempotent as possible so that they do not cause problems no matter how many times a message is handled.

As mentioned in the previous section, putting a GUID inside each message is an obvious solution. Consumers store handled messages in a durable storage, and each consumer must verify whether a message has been handled before handling it.

In addition, consumers must avoid duplicate updates when updating the storage by using transactions.

Let’s take MySQL as an example.

1
2
3
4
5

START TRANSACTION;  
$RET = UPDATE event SET handled = true WHERE msg_id = 123 AND handled = false;  
if $RET == 1:  
    Do whatever you want;  
COMMIT;

Please notice in the first line after the transaction, in addition to msg_id in the WHERE clause we must also add handled = false, this is one of the ways to avoid MySQL having racing conditions. If we don’t add this check condition, it is still possible to process the same message twice under race conditions.

There is a detailed explanation of how MySQL handles race conditions in my previous article, if you want to know the details you can refer to the following.

Nevertheless, it is not easy to make the message idempotent. In addition to the consumer, it often requires the cooperation of the producer, which adds to the coupling and complexity.

Event Sourcing

Another approach is to use event sourcing. To implement event sourcing, the message queue must support querying message logs, such as Kafka.

Event sourcing means that the current state of the consumer is generated by replaying all events, so any event handling failure can be recovered by simply replaying the event from the beginning.

However, this approach creates two problems.

1. The efficiency of replaying from scratch is not good.
2. It is difficult to get a list of several current states.

To solve the first problem, a checkpoint mechanism is usually used. Take snapshots of the states and record offsets at regular time intervals or for each batch of messages handled.

When a problem occurs, just restore the last snapshot and replay the last offset to the latest data. Through the checkpoint mechanism, we can greatly improve the replay efficiency.

The second problem is usually solved by using CQRS, so that a variety of different read requirements have a snapshot available for reference. For more information on how to evolve the CQRS architecture, please refer to my previous article.

To sum up, in order to achieve an exactly-once guarantee by using event sourcing, the consumer must be able to record the handled states and offsets; when each error occurs, the consumer discards the current result and tries to replay it from the previous state snapshot to get the correct result.

Sink Perspective

Sink means downstream of the message. In fact, the sink perspective is a special case of the customer perspective.

Suppose the sink is a database, the consumer itself can already achieve exactly-once guarantee through the event sourcing.

But this is not enough, because in the process of recovering the consumer will propagate the event to the next again, resulting in downstream receiving duplicate results of the same event.

In the above process, the consumer itself can guarantee exactly-once by restoring the state, but the downstream database has been written twice, which is still a problem.

Honestly, there is no silver bullet in sink that can completely achieve exactly-once guarantee without side effects.

Let’s check Solution 1, we only output the accumulated results after the consumer has successfully created a checkpoint.

Since consumers are guaranteed exactly-once through the checkpoint mechanism, we only output the results when the checkpoint is created, thus ensuring consistency between consumer and sink. It is important to make the checkpoint creation and output be transactional.

However, Solution 1 has an obvious drawback. The message handling loses its real-time nature and generates spikes in the database. Because the checkpoints are created in batches, not for each message, i.e., there are no current results in the database. In addition, the output process is also batch, which results in a large number of writes to the database at regular intervals.

So, let’s see Solution 2, which makes the database update idempotent.

It is straightforward to simply make each database update have a primary key and overwrite the rows with the same primary key.

However, Solution 2 also has drawbacks. Firstly, it is not easy to define the primary key for each update. Secondly, when a consumer fails to handle and starts a replay, the saved results may time travel and be reverted back to the past then updated again.

This inconsistency depends on how often the checkpoints are created and how efficiently they are replayed, which can take up to a couple of minutes.

Both Solution 1 and Solution 2 have their potential risks, but if I had to choose, I would prefer Solution 2 more.

Conclusion

In this article, we take a closer look at the semantics of exactly-once and understand how end-to-end exactly-once can be implemented.

Starting with the message producer at the beginning, the message is sent and enters the message queue, then is read and handled by the consumer, and finally the result is output downstream.

I have to say exactly-once guarantee is a very difficult holy grail to achieve, and it has to be done correctly at every stage of the message flow in order to fully achieve end-to-end exactly-once guarantee.

I believe there must be some original designs besides listed in this article that I have never noticed, if you have your own implementation, please feel free to share with me.

Originally published on Medium

Playing PyFlink in a Nutshell

Understanding the power of PyFlink

Last week, we introduced how to build a PyFlink experiment environment, and today we will use that experimental environment to explore the possibilities of PyFlink.

PyFlink is a general purpose streaming framework and abstracts streaming processing into four levels.

1. SQL
2. Table API
3. DataStream
4. Stateful Stream Processing

The closer to the bottom the more flexibility is available, but also requiring writing more code. I would like to be able to do almost everything with PyFlink, so let’s get started with the basic concepts of PyFlink development from a DataStream perspective.

This article will introduce a few key points of PyFlink development with simple descriptions and examples, but will not mention the implementation details of Flink.

DataStream Concept

The development of DataStream will follow the following process.

Basically, we get streaming data from a source, process it, and output it to somewhere.

This is expressed in PyFlink as follows.

1
2
3

ds = env.add_source(kafka_consumer)  
ds = ds.map(transform, output_type=output_type_info)  
ds.add_sink(kafka_producer)

Source and sink are easy to understand, but the key is what processing can be used?

In the official document there is a list of all available operations.
https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/datastream/operators/overview/

For the above example, we used map.

There is also an example of the operation in the Flink artifact, the code is placed at ./examples/python/datastream/basic_operations.py。

What is the difference between map and flat_map?

There are two similar operations in the operation list, map and flat_map, what is the difference between these two operations?

The difference is in the number of generated outputs.

In the case of map, an input event generates one and only one output event; on the other hand, flat_map can generate zero to many output events.

Let’s use the actual code as an example.

1
2
3
4
5
6
7
8
9

def map_transform(i: int):  
  return i * i  
    
def flat_map_transform(i: int):  
  for idx in range(i):  
    yield idx  
      
ds.map(map_transform, output_type=Types.INT())  
ds.flat_map(flat_map_transform, output_type=Types.INT())

In this example, map squares all the input integers and passes them out, one input corresponds to one output. However, flat_map outputs a series of events, and the number of output events is determined by the input events.

If the input is 0, then yield of flat_map will not be triggered, and nothing will be generated.

State

State is the best feature of Flink.

Although we have various operations available, many of them actually produce results based on previous events. How do we keep the previous events? This is where State comes in.

State can be considered as an internal storage in order to persist data, and the size of State is the summary of every node’s memory.

Nevertheless, State can be persisted in a durable storage like RocksDB to gain more scalability.

1

from pyflink.datastream import StreamExecutionEnvironment, EmbeddedRocksDBStateBackend

1
2

env = StreamExecutionEnvironment.get_execution_environment()  
env.set_state_backend(EmbeddedRocksDBStateBackend())

To use State in Flink framework, there are two key points worth noting.

1. State can only be used in “Keyed Data Stream”.
2. State is based on operations and not able to share with others.

All available States and the reference are listing below.
https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/datastream/fault-tolerance/state/

In fact, an example at ./examples/python/datastream/state_access.py also provides a good demonstration.

Connect (Shared State)

As mentioned in the previous section, states are operation-based and cannot be shared, but sometimes we indeed need to combine two different streaming states, so what should we do?

Fortunately, Flink provides connect to enable us to share the states of different streams within the same job.

By using connect, we can combine different streams and use the same operation, so that we can share the same operation state.

To be more concrete, let me provide a practical example. There are two streams.

• Stream 1 provides a mapping between the item identifiers and the item names. When the item name changes, an event (item_id, item_name) is sent into the stream, so we just need to save the latest status.
• Stream 2 is the transaction history, including which item was sold and the number of items ordered.

What we want to do is, when any purchase is entered, then we have to sum it up and append the latest item name to it.

This is the classic streaming enrichment pattern, and I explained the enrichment design pattern in detail in my previous article.

Here is the full program example.

In flat_map1 the stream 1 is handled, that is to say, the mapping of item number and item name is maintained, so this stream does not need to generate output events.

The core of the whole application is in flat_map2. We take the accumulated quantity from self.cnt_state and not only add the new quantity but also update it back to the state. Then, in the output process, we take the corresponding name from self.state and finally we output the enriched events.

Conclusion

In the last example, we demonstrate the operation, state, and merging of streams.

From this example, we can easily understand that Flink can do anything we want as long as we write the program correctly.

We will continue to do some experiments on stream processing and will continue to publish any further updates if there are any.

Originally published on Medium

Playing PyFlink from Scratch

Understanding PyFlink 1.15.2 on Kubernetes

I am trying to play with PyFlink recently, but there is not much information available on the Internet, and most of the information is a bit outdated. So I’ll document how I set up the environment and actually experiment with PyFlink.

Before starting, let’s set the experiment goals.

1. To run PyFlink instead of Flink, because I can’t code in JAVA.
2. To run on K8s, not docker or standalone.
3. Use the latest version of Flink, 1.15.2.

I briefly describe the entire experimental process.

1. Create a K8s cluster.
2. Deploy PyFlink cluster.
3. Upload the job from the local machine.
4. Check the job status in the dashboard.

Now, I believe you all understand the experimental process, so let’s start preparing the preconditions.

• Do not use M1 chips. The current stable version 1.15.2 does not support M1 chips, so installing PyFlink on M1 would be suffering.
• JAVA 11
• Python 3.6–3.8

Once all is ready, let’s get started!

Create a K8s cluster

My personal preference is to use K3s to create clusters rather than minikube.

There are several reasons.

1. It is easier to share the environment. My experimental environment is on EC2, so anyone who knows the location can use it.
2. It is more resource efficient. minikube has high system requirements, while K3s itself is designed for IoT devices and is relatively resource efficient.
3. K3s is a regular K8s, there are no special features.

After setting up the cluster according to the official document, some extra settings are required.

Create a namespace.

• kubectl create ns flink

Create a service account.

• kubectl create serviceaccount flink -n flink

Grant authorization for the service account.

• kubectl create clusterrolebinding flink-role-bind --clusterrole=edit --serviceaccount=flink:flink

Build a PyFlink cluster

To build a PyFlink cluster we need to prepare the container image to support PyFlink first as follows.

https://hub.docker.com/r/wirelessr/pyflink

The Dockerfile to build the container is also available in Dockerhub.

Next, download the latest stable version of flink artifact locally.

https://www.apache.org/dyn/closer.lua/flink/flink-1.15.2/flink-1.15.2-bin-scala_2.12.tgz

After decompression we have to add some fat jar that is required but not packaged into it.

Download bcprov-jdk15on and bcpkix-jdk15on jar files at

• https://mvnrepository.com/artifact/org.bouncycastle/bcprov-jdk15on
• https://mvnrepository.com/artifact/org.bouncycastle/bcpkix-jdk15on

Then move them to the folder

./flink-1.15.2/lib

When everything is ready, we can run the PyFlink cluster on K8s.

1
2
3
4
5
6
7
8
9
10
11
12

./bin/kubernetes-session.sh \  
 -Dkubernetes.namespace=flink \  
 -Dkubernetes.jobmanager.service-account=flink \  
 -Dkubernetes.rest-service.exposed.type=NodePort \  
 -Dkubernetes.cluster-id=flink-cluster \  
 -Dkubernetes.jobmanager.cpu=0.2 \  
 -Djobmanager.memory.process.size=1024m \  
 -Dresourcemanager.taskmanager-timeout=3600000 \  
 -Dkubernetes.taskmanager.cpu=0.2 \  
 -Dtaskmanager.memory.process.size=1024m \  
 -Dtaskmanager.numberOfTaskSlots=1 \  
 -Dkubernetes.container.image=wirelessr/pyflink:1.15.2-scala_2.12-python_3.7

There are two key points noteworthy.

1. rest-service.exposed.type=NodePort. In order to allow users to access the PyFlink cluster, and to simplify the experiment process, we choose to use NodePort as the external interface.
2. We must specify the container.image that can support PyFlink, in this case, the Dockerhub mentioned earlier.

By using kubectl get svc -n flink we can know which port the dashboard (flink-cluster-rest) is running on. This dashboard location will also be the entry point for the jobs we want to submit later.

Submit a job

Most of the Flink examples will use examples/batch/WordCount.jar as the first Hello World.

So we can also use this Jar file to test whether Flink can run correctly.

1
2
3

./bin/flink run \  
  -m ${flink-cluster-rest}:${port} \  
  examples/batch/WordCount.jar

After submitting, we can check the result in the dashboard, normally, it will be executed successfully. But this example is JAVA-based, so we’ll proceed with a Python-based example to verify PyFlink.

Before we start, we need to install the PyFlink package locally.

pip3 install apache-flink==1.15.2

If we are not using M1 chip, this command should install successfully without any problems.

Same as the WordCount example, Python also has a corresponding WordCount example in examples/python/table/word_count.py.

However, the Python example needs to be modified a bit to work properly. Find this comment line：

remove .wait if submitting to a remote cluster

Follow the instructions to remove the .wait above, and then we try to submit the Python job.

1
2
3

./bin/flink run \  
  -m ${flink-cluster-rest}:${port} \  
  -py ./examples/python/table/word_count.py

Returning to the dashboard should show the job is also performing properly.

Conclusion

There is very little information about PyFLink, both in terms of environment installation and coding reference, so I hope this article will be helpful to you.

However, this article does not explain the details of Flink cluster and how to develop PyFlink in detail, I’ll leave those details for a future article.

Originally published on Medium

Building Apache Pinot and Presto

Working on streaming database step by step

Recently, we have been surveying some streaming database solutions and the primary target is Apache Pinot, which fits our needs from the description and is therefore the primary target.

Apache Pinot, a real-time distributed OLAP datastore, purpose-built for low-latency high throughput analytics, perfect for user-facing analytical workloads.

However, some of the official documents are actually missing, so I found some information on the Internet and finally was able to do some experiments. These processes were not documented at that time, so I am not sure what reference materials were actually used.

Currently, the entire experimental infrastructure is built with docker-compose, and the complete codes in Github repository is as follows.
https://github.com/wirelessr/pinot-plus-presto

Apache Pinot

To explain the core components first, let’s take a look at docker-compose.yml.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

services:  
  zookeeper:  
    image: zookeeper:3.5.6  
      
  pinot-controller:  
    image: apachepinot/pinot:0.9.3  
    ports:  
      - "9000:9000"  
      - "8888"  
    environment:  
      JAVA_OPTS: "-javaagent:/opt/pinot/etc/jmx_prometheus_javaagent/jmx_prometheus_javaagent-0.12.0.jar=8888:/opt/pinot/etc/jmx_prometheus_javaagent/configs/pinot.yml -Dplugins.dir=/opt/pinot/plugins -Xms1G -Xmx4G -XX:+UseG1GC -XX:MaxGCPauseMillis=200 -Xloggc:gc-pinot-controller.log"  
      
  pinot-broker:  
    image: apachepinot/pinot:0.9.3  
    ports:  
      - "8099:8099"  
      - "8888"  
    environment:  
      JAVA_OPTS: "-javaagent:/opt/pinot/etc/jmx_prometheus_javaagent/jmx_prometheus_javaagent-0.12.0.jar=8888:/opt/pinot/etc/jmx_prometheus_javaagent/configs/pinot.yml -Dplugins.dir=/opt/pinot/plugins -Xms4G -Xmx4G -XX:+UseG1GC -XX:MaxGCPauseMillis=200 -Xloggc:gc-pinot-broker.log"  
      
  pinot-server:  
    image: apachepinot/pinot:0.9.3  
    ports:  
      - "8098:8098"  
      - "8888"  
    environment:  
      JAVA_OPTS: "-javaagent:/opt/pinot/etc/jmx_prometheus_javaagent/jmx_prometheus_javaagent-0.12.0.jar=8888:/opt/pinot/etc/jmx_prometheus_javaagent/configs/pinot.yml -Dplugins.dir=/opt/pinot/plugins -Xms4G -Xmx16G -XX:+UseG1GC -XX:MaxGCPauseMillis=200 -Xloggc:gc-pinot-server.log"

The above are the four basic services needed to construct a Pinot, most of the same as the official document are omitted and only the parts I modified are listed. By the way, I have changed the image version to the new one, but let’s keep the official version in the example.

The main changes are to the ports and environment of the three Pinot services, in order to make Pinot’s metrics available by Prometheus.

Therefore, ports opens an additional 8888 and adds an additional javaagent parameter to JAVA_OPTS. The purpose of the javaagent is to enable the original Pinot JMX metrics to be web-based and exposed to Prometheus.

So far, the jmx_prometheus_javaagent.jar in the official document does not include the version number, but I could not find the corresponding file from the official container, so I have to use the jmx_prometheus_javaagent-0.12.0.jar instead.

Monitoring

In order to observe the state of the services, we also need to build Prometheus and Grafana.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

prometheus:  
    image: prom/prometheus  
    container_name: monitoring-prometheus  
    restart: unless-stopped  
    volumes:  
    - monitoring-prometheus-data-1:/prometheus  
    - ./volumes/prometheus/prometheus.yml:/etc/prometheus/prometheus.yml  
    ports:  
    - "9090:9090"  
  grafana:  
    image: grafana/grafana  
    volumes:  
    - ./volumes/grafana/provisioning:/etc/grafana/provisioning  
    - ./volumes/grafana/dashboards:/var/lib/grafana/dashboards  
    ports:  
    - "3000:3000"  
    environment:  
      GF_SERVER_ROOT_URL: https://localhost:3000  
      GF_SECURITY_ADMIN_PASSWORD: password

In volumes we define the required configuration files for these two services.

• Prometheus needs to know where to fetch the metrics from, so it needs to set the FQDNs of the three Pinot services.
• Grafana requires data source settings and a Pinot-specific dashboard.

After running docker-compose, we can view the dashboard from https://localhost:3000. The account and password are simply admin and password.

Apache Presto

The last service is Presto.

Why do we need Presto? Because Pinot supports SQL, but in fact Pinot does not support JOIN. If there is a need to merge two tables, it must be provided by an external SQL engine.

Presto is a SQL engine that supports ANSI SQL and multiple data sources, including Pinot of course, so I chose to use Presto in my experiment.

1
2
3
4
5

presto-coordinator:  
    image: apachepinot/pinot-presto  
    restart: unless-stopped  
    ports:  
    - "18080:8080"

Presto has two roles, coordinator and worker. In the experimental environment, we simply use the built-in worker of the coordinator.

The official container apachepinot/pinot-presto is the coordinator by default without any settings, and already uses the FQDN pinot-controller:9000, so there is no need to change anything, we can use it directly.

Conclusion

After cloning the Github repository, we can run docker-compose up -d directly to get the experimental environment up and running well.

The whole experimental environment contains three key points.

1. Apache Pinot: the streaming database itself.
2. Prometheus and Grafana: the monitoring system that emulates the production environment.
3. Apache Presto: a query capability that is not available in Pinot.

By the way, the official document does not introduce how to use Kafka with security, only Kafka without security as an example.

In the case of Confluent, Kafka has built-in SASL_SSL, I refer to this article to set up Kafka’s account and password.
https://dev.startree.ai/docs/pinot/recipes/kafka-sasl

This experimental environment may be expanded for some of my needs, such as adding Presto workers.

Originally published on Medium

Start from a small step

Last time we talked about the evolution of the data infrastructure, the whole evolution process is as follows.

• Stage 0: None
• Stage 1: Batching
• Stage 2: Streaming
• Stage 3: Integrated
• Stage 4: Automation
• Stage 5: Decentralization

But remember? The evolution from stage 1 to stage 2 was a bit weird in that we just dropped the batch processing we were using and built a new stream processing.

In fact, a product that is already on production environment not make such a drastic change, but would choose an incremental migration process instead.

Therefore, in this article, we will explain in detail how the migration from stage 1 to stage 2 will work in practice.

Stage 1: Batching

Let’s review the batch architecture again.

Under this architecture, we have a batch processing role responsible for archiving data from various services to the data warehouse. And batch processing also performs data pre-processing to generate structured data for accelerated analysis purposes.

However, in order to support real-time analytics, we knew from the previous article that we had to introduce a streaming architecture, but at the same time we wanted to leave the original analytics functionality as is.

In other words, it was necessary to add additional real-time analytics capabilities while maintaining the existing architecture and functionality.

So, let’s start trying to add streaming architecture.

Stage 1.5: Batching + Streaming

In the previous article, we took an aggressive approach by simply dropping batch processing and replacing it with streaming. This would make the architecture simple, but comes at the cost of a complete rewrite of all the analysis, since the data from streaming is not in the same format as the data from batch processing.

Therefore, we add an extra 1.5 stage to allow batch and streaming to coexist for a while until we could change the analysis framework to use streaming data sources for all subsequent analyses.

The advantage of this is of course that we don’t need to do a Big Bang.

As Martin Fowler reportedly said, “the only thing a Big Bang rewrite guarantees is a Big Bang!”.

However, the drawback is obvious, when there is any requirement for modification, we have to implement both batch processing and streaming processing, which greatly increases the difficulty of maintenance.

Therefore, although there is a 1.5 stage, it is important to realize that this is only a transition period and should not be considered as a solution.

Stage 2: Streaming

Once we have a transition period, we can evolve more smoothly from batch processing to streaming processing. In the previous stage, we will gradually move all the functionality from batch processing to streaming processing.

At the end, it’s just easier to take away the batch processing.

But as I stated in my previous article, batch processing still has its value, so even if it is taken out of the main data pipeline, we will still have batch processing there.

It will just change from being the primary data producer to a secondary one, regularly reorganizing the data in the data warehouse to improve data utilization. Of course, batch processing can also be used to move out-of-date data from the data warehouse to cold storage, further saving storage costs.

Conclusion

The whole evolutionary process should be smooth and gradual.

In order to avoid a Big Bang, it is a difficult choice to move from one architecture to another. Each migration should have as small impact as possible, but at the same time, it should not be too small to make the pain period too long.

Overall, every organization has a different tempo. For an organization with sufficient human resources, this forward movement can be aggressive and fast, or on the contrary, measured.

But in any case, it is important to understand the dilemma and come up with a reasonable enough approach and, more importantly, a complete migration plan.

This planning process is actually as important as the architecture change.

Originally published on Medium

From Monolith to Self-service Platform

All systems start as a small monolith. In the beginning, when resources and manpower are not sufficient, the monolith is the choice we have to make, even the data infrastructure is no exception.

But as requirements increase, there are more and more scenarios that cannot be achieved by the current architecture, and the system must therefore evolve. Each time the system evolves, it is to solve the problems encountered, so it is necessary to understand the different aspects that need to be considered, and to use the most efficient engineering methods to achieve the goal.

In this article, we will still start with a monolith, as we have done before.

• Shift from Monolith to CQRS

But this time, our goal is not to serve a production environment, but to provide the data infrastructure behind all production environments.

A data infrastructure is a “place” where all kinds of data are stored, either structured data or time series data or even raw data. The purpose of this big data (and they are really big) is to provide material for data analysis, business intelligence or machine learning.

In addition to internal uses, there may also be user-facing functions, for example, a list of recommended products on an e-commerce website. The recommendation function is the most commonly used customer scenario for big data, such as recommended products in e-commerce websites and recommended videos in video platforms like Youtube. All of these are the result of analyzing, aggregating, and compiling after storing various user actions.

After knowing today’s topic, let’s get to work.

Monolithic architecture

Let’s begin with a monolithic architecture.

Whatever the service is, a typical monolithic architecture is a service paired with a database. All data analysis, business intelligence and machine learning run directly on this database.

This is the simplest practice, but it is also the beginning of all products.

Why do we show it in monolithic style? Wouldn’t it be better to do more?

Well, if the product dies before it is popular, then the monolith will be enough to maintain full features. In order to avoid over-engineering and the investment cost getting exploded, we prefer to start with the easiest proof-of-concept monoliths.

Of course, this architecture is sure to encounter problems. There are three common problems.

1. The analysis affects the production performance.
2. If the microservices are introduced, running analysis on each microservice’s database is suffered.
3. It is very difficult to perform cross-service analysis.

Various analyses inevitably take up a lot of database resources, so customers directly feel the poor performance during that time.

In addition, when product requirements increase, we usually adopt microservices for rapid iteration, which means many service and database pairings are created. In order to be able to analyze various product requirements, it is necessary to build several analysis frameworks on every database.

Moreover, even if each service can be analyzed individually, we cannot easily obtain a global view because the analytics are independent of each other.

Therefore it is necessary to perform the first evolution.

Batch processing

We have two goals, firstly to have independent data storage so it does not directly affect the production environment, and secondly to be able to perform cross-service analysis.

To achieve these goals, we set up a centralized data warehouse so that all data can be stored in one place. All analyses were migrated from the databases of each service to the data warehouse.

We also need a batch processing role to collect data and even pre-process data periodically so that the data warehouse contains not only raw data but also structured data to speed up the analysis.

For batch processing, I recommend using Apache Airflow, which is easy to manage and easy to script for various DAGs to address the needs of multiple batch processing scenarios.

Such an architecture is simple but powerful enough to handle a wide range of analysis and reporting scenarios. Nevertheless, there are two drawbacks that cannot be easily solved.

1. lack of real time
2. lack of schema

Due to the periodic batch tasks, we can’t do real-time analysis. If we run the task once an hour, then the data we see in the data warehouse is a snapshot of the previous hour’s data.

In addition, batch tasks require knowledge of the data schema of each service in order to get the data correctly and save it to the corresponding warehouse table. Assuming our data warehouse is GCP BigQuery, the schema in the warehouse table also needs to be created and modified manually.

When the number of services is very large, just knowing the schema can take a lot of time, not to mention managing it.

So, let’s do a second evolution.

Stream processing

To solve the real-time problem, the most straightforward idea is to use a streaming architecture.

First, all the database changes of the service are captured, then the changes are sent to the stream, and finally the stream is archived to the warehouse.

But is there a way to solve the schema problem? The answer is, yes, through KCBQ.

Let’s take a closer look at the internal architecture of streaming.

We use Debezium to capture changes to each database and send the streams to Kafka, and later KCBQ subscribes to the Kafka streams and archives them to BigQuery.

It is worth noting that KCBQ can automatically update the internal schema of BigQuery (setting needs to be enabled, and AVRO format must be used).

At this stage, we have a complete data set for analysis, and we have enough data to process for all kinds of data analysis, both real time and batch. However, in the previous stage, we were able to accelerate the analysis by batching pre-processing to generate structured data, and we still want to retain this capability.

In addition, although the data analysis is working properly, we are currently unable to support external customers to access the data. The main reason is the slow response time of data warehouse, which is a big problem when we need to interact with external customers. For example, if a recommendation list takes 5 minutes to be generated, the effectiveness of the recommendation will be close to zero.

Furthermore, when the number of features to be provided to external customers increases, data warehouse can be difficult to handle. For example, search is a common feature, but the search performance of data warehouse is very poor.

Therefore, we have to integrate batch processing and streaming processing so that this data infrastructure can flexibly support various customer-oriented functions.

Integrated architecture

We already have a batch architecture and a streaming architecture, now let’s integrate them properly.

We still archive the changes from each database to the data warehouse through streaming, but we periodically use batch processing to pre-process the raw data for data analysis.

At the same time, the streaming framework not only archives, but also aggregates the streams to generate real-time customer-facing data storage. For example, ElasticSearch is often used in search contexts, while the event-sourced view is used to provide customer-facing functions, e.g., recommendation lists.

But these functions usually also require fact tables, which can be migrated from data warehouses if they are static, otherwise, they need to be implemented through stream enrichment.

So let’s further have a look at the internals of streaming architecture.

The framework is basically the same as the previous one, but with a new role, StreamAPI.

There is no specific framework specified in StreamAPI, even if using Kafka’s native library is fine.

But if using the stream processing framework can save a lot of implementation efforts, such as data partitioning, fault tolerance, state persistence, and so on.

Therefore, I still recommend using a streaming framework such as Apache Flink or Apache Kafka Streams.

At this point, the entire data infrastructure is fully functional. Whether it is for internal data analysis or for external customers, this framework can meet the requirements.

However, this architecture doesn’t mean that it’s done once and for all.

The next problem is the management problem. So far, the most common task we need to do is not feature development, but the configuration and operation of so many middleware. For each new service, we need to build Debezium, add Kafka topic, configure the streaming framework, and determine the final sink. This process also applies to a new feature request.

In addition, managing the permissions between all data stores is also a problem. When a new person comes into the organization, each participant in the framework has to grant some permissions. Similarly, when a new service or feature is launched, the permissions between services need to be managed.

What’s more, when all the data is dumped into the data warehouse, we have to encrypt or mask the PII (personally identifiable information) to be compliant with legal requirements, such as GDPR.

These tasks rely on a lot of manual work or semi-automatic scripts, which undoubtedly add a heavy load to a busy organization.

Automation (Self-service Platform)

Therefore, it is necessary to establish a fully automated mechanism to solve all management issues of the data infrastructure.

Of course, this automation mechanism should ideally be self-service, and the most successful example I have seen is Netflix’s Keystone.

This automation platform involves very deep technical details, not an article can simply explain, so I only provide the features of this platform should have, as for the implementation details will depend on the practice of each organization.

At the infrastructure level, in order to automate the establishment of settings and monitoring, it is necessary to achieve fully automated deployment through IaC (Infrastructure as Code).

In order to address the issue of data ownership, there are three permission-related controls that should also be included in the IaC as follows.

• Identity Access Management, IAM
• Role Based Access Control, RBAC
• Access Control List, ACL

Privacy issues require an easy-to-use interface that can be configured to determine what data must be masked. Data Loss Prevention, aka DLP, is also a very popular data privacy topic nowadays.

As you can see from the above description, there are a lot of implementation details to achieve a fully automated management platform, and there are no shortcuts — implementation, implementation and lots of implementation.

Conclusion

This article is about the evolution of a monolithic architecture into a fully automated and fully functional data platform through continuous problem solving.

However, there are still a lot of detailed technical selections in the evolution process, such as what data storage to use, what batch processing, and what real-time processing are all worthy of attention.

Such an evolution requires a lot of human resources and time, so it is unlikely to be achieved in one step. If and only if we encounter a problem that cannot be solved, we will be forced to move to the next stage.

In fact, even if we achieve an automated platform, the evolution is still not over.

When all kinds of business requirements and all kinds of products and services have to rely on a single data engineer department, then human resources will encounter a bottleneck. How to solve it?

Since we have a fully automated management platform, we can let each product department maintain its own data warehouse and data pipeline. The data engineer becomes the provider of the platform solution, not the owner of the data.

The Data Mesh is born.

However, there are many different approaches to the Data Mesh implementation, and there is no one-size-fits-all architecture pattern.

This is all about the evolution of data infrastructure. If you have any special practices, please feel free to share them with me.

Originally published on Medium

Archiving MongoDB Changes to AWS S3

Understanding Hot-Cold Separation

In this article, we are going to introduce how to archive the audit logs stored in MongoDB to AWS S3, but before we start, let’s understand why we need it and the meaning of the need.

Audit logs are records of user actions on the system, including signing in and out, modifying data, enabling a feature, and so on. These records are kept as part of the system security, and recent data is made available for users to view. Users may be interested in the last three months of related operations, but not in records older than that.

Therefore, data that is older than three months is classified as hot data, while data that is older than three months gradually becomes warmer, i.e., rarely accessed, perhaps sometimes still. Data older than six months will be completely cold and will not be accessed except for data analysis and internal system auditing, and users will not care about the existence of such data.

That is to say, if these cold data are still in the original database, it will consume a lot of resources, including hard disk space and indexes. So, we need a way to archive these data to a lower cost storage space, but we still need to make the cold data directly available when they are to be analyzed, instead of doing further migration.

So, today’s title tells you that we will archive the data in MongoDB to AWS S3.

There are several advantages to S3, first of all, the data in AWS S3 can be easily analyzed and SQL-compatible queries can be performed through AWS Athena.

In addition, S3 is also a storage option supported by many data analytics platforms, such as Apache Spark.

Furthermore, storage cost of S3 is very low. Moreover, if it is determined that the data in S3 is for preservation purposes only and not for query purposes, it can be easily transferred to much lower cost storage, such as AWS Glacier.

Hot-Cold Separation on MongoDB

Now let’s review the requirement again.

In order to reduce the storage cost, we need to move the past audit logs from MongoDB to AWS S3. In addition, we want this to be fully automated, and I don’t believe anyone wants to manually migrate data periodically.

So what’s the best way to do this? Yes, through streaming.

We have three things to do.

1. Pass the events from the new audit log through MongoDB’s CDC (and ignore the delete events).
2. The stream processor archives the stream events into AWS S3.
3. Use MongoDB’s TTL index to automatically delete expired data when adding audit logs.

By the way, the MongoDB mentioned here is not a self-hosted database, but a managed platform through Atlas. However, the same approach still applies to self-hosted databases, just with a little more setup and implementation than the managed platform.

After knowing what to do, let’s work out a solution.

Approach Details

Then, let’s explain the whole solution in detail and provide some key points worth discussing.

The front services and databases are original to the architecture, only the back of the streaming chain is new. The entire chain ends with our destination, AWS S3.

The first thing to do is to turn on Atlas’ built-in feature, EventBridge Triggers, and since it is a built-in feature, it only requires a few simple settings following the official documentation. There are a few key points to note here, first we only need to send INSERT events, and we need to turn on the option to send full documents so that we can save the complete audit data.

Next, we need to add a new rule to EventBridge’s third-party rules: all incoming events are sent to Kinesis, so that the source of the stream is established.

In order to fully automate S3 archiving and save transmission costs as much as possible, we add Kinesis Firehose as the stream aggregator in the middle of Kinesis and S3. Through Firehose, we can aggregate a stream of events into a batch, and write to S3 in a batch process to save S3’s transmission cost.

In this way, a fully automatic archiving is accomplished. All audit logs written to MongoDB will be archived into AWS S3 through streams.

Let’s discuss a few topics.

Why not just let the service write S3 directly?

There are two reasons.

The first is that we can achieve the purpose without changing the original application by such modification.

The other reason is that audit logs usually require transactional consistency, and it is difficult to ensure that MongoDB and AWS S3 are consistent when the primary database is MongoDB. What if writing MongoDB succeeds, but writing S3 fails?

Why Kinesis and Firehose?

In fact, we can retain one of them, if we want to keep only one I propose to keep Firehose, because we can reduce transmission costs through batch processing.

But Kinesis has a benefit, if we later want to introduce event-driven architecture or streaming processing architecture, Kinesis can provide a very good extensibility.

Does MongoDB’s TTL index affect database performance?

Yes, it does, but the impact is limited in this context. This is because we will only keep the data for a certain period of time, for example, three months.

Therefore, the amount of data is limited and the impact on the database is manageable.

Conclusion

Streaming can empower the scalability of the entire system design.

Whether it’s through database CDC or events from microservices, it can effectively improve system availability and decouple business logic from data. That’s why I’ve been sharing the streaming architecture for the last few weeks.

This time we experienced the convenience of streaming through the managed platform provided by AWS, a fully automated archiving system. Without the help of streaming, we would have to find a way to activate some scheduled mechanisms, such as a crontab, and get up once a day to check what audit logs needs to be moved, and to do large-scale data migration.

If we use such a batch design, we will immediately encounter a problem, lack of real time. When we do data migration through periodic tasks, there is a data lag, for example, one day. Any data analysis performed on S3 will not get immediate results, but rather a snapshot of some point in time in the past, which is the biggest problem with batching.

But we haven’t actually introduced some of the famous streaming framework yet, so maybe we’ll have a chance to do that in the future, but rather than actually operating streaming, I feel I’d share some more topics related to architecture and system design.

Originally published on Medium

Reducing Database Loading

Three approaches to improve database elasticity

Last week we talked about the difference between scalability and elasticity, and the concept of solutions was mentioned at a high level. In this article, let’s take a look at some practical solutions.

First, if you haven’t read the last article, I suggest you read it first. This time we are solving the elasticity problem rather than scalability. As mentioned before, the two most common solutions to the elasticity problem are caching and messaging.

Therefore, in this article, we will list some common practical solutions and briefly analyze the advantages and disadvantages. Let’s get started.

Read-aside Cache

When talking about caching, the most commonly used mechanism is the read-aside cache.

The advantage is that it is very easy to understand and implement, and already provides a good consistency level. I’ve explained how it works in my previous article, but let’s review it again.

The read mechanism of the read-aside cache is very simple: first try to get the data from the cache, and if you cannot get it, then get it from the database instead, and save it back to the cache. In this way, the next request for the same data can be taken directly from the cache.

How to ensure that the cached data is up-to-date (consistency)?

After updating the database, the corresponding records of the cache should be cleared immediately. When the same request comes in next time, it will be taken from the database first and the latest result will be written back to the cache.

It looks like both read and write are perfect, so the read-aside cache should have a high degree of consistency! In fact, not really.

According to the above sequential diagram, we can find that both A and B are behaving correctly, but put together, they make an error, and B gets inconsistent data. This is the first problem.

The second type of error is more intuitive, when A wants to update the data, A is killed after finishing the database update, probably due to bugs or application upgrade and so on. Then the data in the cache will remain inconsistent for a long time, until the next update or timeout.

The third error is similar to the first one in that the individual actions of A and B are correct, but put together they are wrong. This inconsistent cache data will also remain for a while until the next update.

These details have been explained in my previous article, if you want to know more can also be used as a reference.

In fact, problems 1 and 3 can be greatly mitigated by modifying the implementation of the application. In the first problem, when A finished writing the data, don’t do anything else, then clean up the cache quickly to reduce the possibility of being touched by B.

The same is true for problem 3. When A finishes querying, don’t do too much processing and write back the cache as soon as possible, which can also greatly reduce the chance of occurring after B.

However, there are three potential problems with read-aside caching.

1. If a cache has a large scope of content, it can be difficult to determine when to clean it and to implement it. On the other hand, when many places have to be cleaned up, it may make the cache less effective.
2. Dog pile effect. This refers to when a high-traffic system accesses cache frequently, if it encounters clearing cache or cache expiration, these traffic will still occur on the database. And the database may not be able to hold up.
3. No graceful shutdown. Problems 1 and 3 can be improved by modifying the application, but problem 2 is not always possible, especially when the application fails, there is basically nothing to do.

Denormalization

If we have to take data from many places, it is ideal to put it all together and take it only once.

Conventionally, in order to avoid data duplication in RDBMS, we would create individual tables according to the data type, and then join each other through foreign keys.

But this will cause a problem, if it is an aggregated data must be pulled from each table, in the traffic and data volume of the system is not large, this approach is reasonable, and will have good performance. However, in a big data scenario, such performance is tragic.

The performance of JOIN is not good enough for large data, and it is even worse if it is a subquery.

Therefore, the idea of denormalization is to put all the data together and pull all the required data through a single query. Common practices include Facts Tables, etc., but this article does not focus on data modeling, but on system architecture.

However, the problem is that reading from one place is possible, but what about writing to it? The original tables are already in use by clients, so do we need to find all the clients to modify them? No, it’s too much work!

There are two ways, synchronization and asynchronization.

This is an architecture of synchronization. The following DBs can be treated as databases or tables or even caches.

When the service needs to write data, originally only DB1, DB2 and DB3 need to be written, in order to achieve denormalization, but also need to write NewDB in code, and NewDB is the result of aggregation.

If someone needs to aggregate results, they can pull them directly from NewDB. If the owners of these DBs all belong to the same functional team, then this implementation looks acceptable.

On the contrary, if the team that needs to consolidate the results is another functional team, such an architecture will not work. First, the other functional team has to find all the updates, then they have to make changes within the service, and they have to do it transactionally.

For a cross-functional team, such a workflow is unrealistic and impossible. Hence, asynchronous data replication is introduced.

This architecture should be very common in my previous articles, it is in fact CQRS.

• Shift from Monolith to CQRS
• Solve Performance Bottleneck through CQRS

Nevertheless, there are various practices to implement CQRS, some of which are listed below. The following is a list of common practices from long to short time intervals.

ETL

DBs -> Batching -> Materialized View

This is the most common practice nowadays, and almost all data engineers do it this way. Data is migrated from one place to another through batch processing, whether it is Hadoop or Spark.

Some transformations and business logic may be added in the process. In order to make the data more accessible, if RDBMS is used as the NewDB, then Materialized View may be added. The data is snapshotted by predefined rules, and if the data is updated, it is refreshed to respond to the calculation of new data.

But data migration is a very costly practice, so ETL does not happen all the time, instead, it happens once an hour or even once a day.

Therefore, the data obtained from NewDB will not be the latest, but a snapshot at a certain point in time.

CDC

In order to keep the data as up-to-date as possible, we know that ETL is not enough. Therefore, CDC is another common practice.

DBs -> Debezium -> Kafka -> Streaming DB

When data is updated, it is captured by Debezium, streamed out to Kafka, and finally written to a streaming database.

The streaming database is made available to external users, so the streaming database must be able to support not only streaming writes, but also rapid query capabilities.

There are many mainstream streaming databases, and Apache Pinot is the most popular one recently.

Because CDC streams are updated in real time, external users can get almost real-time data.

Streaming

Another mechanism for real time streaming comes from the service that originally wrote the data.

Service -> Kafka -> Stream Handler -> RocksDB

When the service finishes updating the original database, it actively sends an event to Kafka and Kafka dispatches it.

In such a system, it is most common to use a streaming framework to do the pre-processing of the streams and store them in the database. Common streaming frameworks include Apache Flink or Apache Samza, and of course the recent Kafka Stream.

For the backend database, we choose a state persistence mechanism that can match the streaming processing framework, and RocksDB is the best choice for several common streaming processing frameworks.

In this way, we can also get the latest data, but from the query point of view, the query capability provided by the streaming processing framework is not comparable to streaming database.

The above three are common data replication methods, and it is worth mentioning that the components of each method can be exchanged or even extended to each other, so that different architectures can be built based on requirements.

However, there are potential problems with asynchronous data replication.

1. If there is a wide range of data to be used, it is a big problem to define the data to be replicated. Just as it is difficult to determine the cleanup time for a read-aside cache, it is difficult to define the scope of data replication.
2. In addition, whether it is pre-processing through streaming framework or defining query through streaming database, the complexity is obviously high when it comes to complex data structure and complicated computation.
3. Among these three methods, ETL has the lowest complexity, but ETL loses data real-time.

Asynchronous requests/responses

We all know the traditional request/response model as follows.

Although this is a function call here, it can also be a remote call across services.

After the client sends a request, the server responds to the request through a series of processing, which is the traditional approach. How can we do this asynchronously?

The answer is to make the implementation of the function behind the server asynchronous.

What are the benefits of doing this?

1. Handle events by using event handlers, then we can control the number of handlers to determine the pressure on the backend system or database. In the synchronization model, when a large number of client requests come in, the server will fork threads accordingly and put pressure on the backend system. However, when the number of event handlers is controllable, the pressure on the backend system can be controlled as well.
2. Request dedup. Since all requests from clients are entered into the message queue, event handlers have the ability to perform deduplication according to specific rules.
3. Further, if a streaming framework is used, then the third approach from the previous section can be followed to achieve further denormalization.

However, such an architecture also has drawbacks.

1. Such an architecture must be based on the assumption that users will refresh the page a lot in a short period of time in order to be able to play the benefits of dedup. This architecture does not mean that it is only applicable to this context, but if there is no matching context, then the implementation will be too much pain and too little gain.
2. Implementation overhead. The server must have the ability to wait for asynchronous messages, which is simple to say, but there are many aspects of the implementation. For example, should we use a temporary queue or a permanent queue? How long should I wait? Questions like these need to be carefully considered.
3. Starvation. In order to control the number of concurrent messages, we don’t enable handlers unlimitedly, but this also means that some people can’t wait for processing and time out.

Conclusion

This article describes some approaches to improve the elasticity of a database.

Unlike the approaches to improve scalability, scalability can be achieved through database-oriented approaches such as query optimization, indexing, and data sharding.

However, improving elasticity cannot be accomplished through databases alone, but must be accompanied by architectural adjustments, so this article lists some common approaches.

In fact, there are many more useful practices than that, and this article is aimed at small to medium-sized organizations. With a limited number of people and a limited budget, it is not possible to make large-scale changes, so we can only optimize on the original practices.

Nevertheless, I believe these approaches should provide good insights for further evolution of the system, after all, all methodologies are similar.

Originally published on Medium

Understanding the main difference between scalability and elasticity

In system design, there are two single words are confusing, which are scalability and elasticity. They are so similar that we don’t distinguish them correctly.

However, when we want to solve the issues caused by these two non-functional requirements individually, we need completely different approaches.

Therefore, in order to design the right approach we have to know them and recognize their properties. In this article, we will talk about what scalability and elasticity are and how to treat the symptoms.

Scalability

We often talked about horizontal scalability and vertical scalability, i. e., scale-out, and scale-up.

But what’s good and bad for scalability?

Let’s take a look at an example.

CPU grows with the time. This is normal. As the system grows, the number of users increases, the feature requirements increase, and thus the resource consumption increases as well. In this case, resources refer to CPU, but it can be applied to other resources, such as memory, response time, etc.

So, if this diagram indicates a normal situation, what diagram indicates a problem?

When resource usage does not rise linearly, but exponentially, something is wrong. It could be a bottleneck in usage that results in additional overhead, or a dramatic drop in performance when the amount of data exceeds a certain amount.

On the other hand, what kind of shape represents good scalability?

Similarly, resource usage rises over time, but the rate of increase shrinks, which means that scalability becomes better. The ideal state is a horizontal line, where the resource usage remains constant regardless of the number of users and the increase in feature requirements.

In other words, how to make this line as straight and as close to the horizontal as possible is the issue to be addressed in solving the scalability.

Elasticity

After talking about scalability, let’s look at an example of elasticity.

First, let’s also look at the normal situation of a system.

Resource usage fluctuates dramatically over time, with spikes occurring at certain times. It is easy to understand why this happens. Take an e-commerce site for example, when there is a Black Friday sale or Christmas coming up, the system usage will be much higher than usual. Or a ticketing site, when a top singer is going to have a concert, will also attract a large number of people to come to grab tickets.

So what is good elasticity? It’s about eliminating those spikes, of course.

There are two key points in this diagram.

1. The red line is more stable than the blue line, but the red line is not a horizontal line.
2. The red line is significantly higher than the blue line after the occurrence of a spike.

The reason for the first point is simple, because the system usage is not fixed, so the use of resources will vary with the system usage. But in general, it can still be maintained in a stable state and there will be no significant ups and downs.

The second point, I believe, is not difficult to understand: the reason why we can flatten the blue line as much as possible is not to reject all the user requests, but to postpone the instantaneous usage through “a sort of mechanism”. Therefore, after the spike, there will always be a period of digestion, causing the red line to be higher than the blue line.

You may ask, though, the system can handle so many users without elasticity, so is it still necessary to improve the elasticity to make the effect of the red line?

The answer is, definitely. Why?

Because the highest point of the blue line is the limit of what the system can record, and more rejected user requests cannot enter the system and are not recorded. In the moment of the spike, the system’s maximum capacity is defined there, and all requests exceeding it are rejected.

On the other hand, a more elastic system has a smoother resource usage and can handle more user requests without crashing the system.

The Myth of Autoscaling

Now, we understand the scalability and elasticity. So, can you answer one question?

Ask, does autoscaling address the need for scalability or elasticity?

The answer is, scalability.

First of all, the principle of autoscaling is to track specific metrics by sampling and start scaling when the threshold is exceeded for a period of time.

Therefore, it is already a while after the spike has occurred before scaling can begin. In addition, scaling is not immediately available; even containerized applications have to go through several steps to scale, such as loading the container image and cold-starting it on a new instance. A complex application can take a few minutes from the time it decides to scale to the time it is launched.

As a result, autoscaling is not about elasticity, but about scalability.

More importantly, autoscaling does not improve the whole system. In other words, autoscaling only allows the system to continue to grow with its original scalability, but does not make it better.

One exception is that if elastic requirements can be predicted, then autoscaling can be effectively applied.

For example, if the system receives a large number of requests at the beginning of the workday (e.g., 9:00 a.m.), then the system can be autoscaled by activating it at regular intervals. In this example, setting the time of the autoscaling between 8:30 am and 9:30 am can effectively solve the elasticity problem faced during that time.

However, in general, the autoscaling faces the issue of scalability.

Conclusion

In this article, we introduced scalability and elasticity and explained what it looks like after improvement.

Nevertheless, this article does not describe how to improve scalability, because scalability is a systemic issue and there is no one specific solution that can solve it all at once.

First, bottlenecks must be identified, then the root causes must be figured out, and finally different solutions must be proposed for each bottleneck. This requires an extra investment of time and human resources to enable the system evolution, and autoscaling is a compromise to buy time. By spending additional money, the system is given time to evolve and is not shut down by the inability to scale.

But autoscaling is not a long-term solution, and additional analysis, design and implementation costs are required to fundamentally solve the scalability issue.

On the other hand, how to solve the elasticity issue?

The answer is relatively simple, through caching or messaging. In order to tolerate a large number of requests in a moment, there must be a mechanism to postpone the current spike, and the most commonly used practice is caching.

However, there are many different ways to implement caching, and the balance between caching and consistency is an important issue. I have introduced several caching practices.

1. Consistency between Cache and Database, Part 1
2. Consistency between Cache and Database, Part 2

These are the basic versions of caching, but there are many more advanced caching practices that will be mentioned in subsequent articles.

The purpose of this article is only to explain the two words that are easily confused, and the only way to find a feasible solution among the many approaches is to correctly understand scalability and elasticity.

Originally published on Medium