CT Wu

There are so many tradeoffs, how to carry on?

Last week, an engineer asked me how to make a technical selection. And he provided a real-world example. Their organization wanted to implement an event-driven architecture, and there were three options available right now — MQTT, AMQP, and Kafka — but he didn’t know how to make a decision.

My answer, as always, was “it depends”.

“Everything in software architecture is a tradeoff”. — Mark Richards

Because of the tradeoffs, there will never be a standard answer, in other words, there will never be a silver bullet.

However, we always have to make a decision, what are the tradeoffs? At that moment, I turned back to him and asked.

• Do you understand your requirements completely?
• Do you know what these three techniques are exactly doing?

Take message queues as an example, I have written an article previously on what use cases to consider when choosing a message queue.

• Propagation: Can fan-out be supported?
• Delivery: What kind of delivery guarantee does it have? At least once or exactly once?
• Persistence: Can messages be persisted?
• Consumer group: Does the receiver scale horizontally?

These are the generic requirements for a message queue, and even if you don’t choose a message queue, listing the generic requirements for a technical stack can more or less filter out some options.

Then, after filtering by generic requirements, further filtering could be done based on the specific use cases. For instance.

• Is there a large volume of messages to be written?
• Is there a limit to the bandwidth that can be transferred?
• What kind of security is required?
• What programming languages are expected to be integrated?

To answer questions such as these, in addition to a precise understanding of the requirements, we must also have a clear understanding of the technical stack candidates.

Therefore, architects use a similar thought process when making a technical selection to clarify requirements and learn about the technical stack, and finally make a decision. The decision is not about choosing the latest technique, or just playing around with it as if it works.

Nevertheless, we will always encounter the situation that all needs are well met, after all, nowadays technology is developing rapidly, and many of our needs have already been implemented, so what should we do? Throw the dice?

Wait, you can ask more questions.

• Should we maintain this technical stack ourselves or purchase a managed service?
• Is there a budget limit?
• How do we monitor it, whether we operate it ourselves or buy a managed service?
• Do we have any experience in using it?

When there is no more question to ask to filter, then the last thing to do is to choose what we love and love what we choose, and not to waver from one option to another.

Any technical stack from finalization to production environment must go through time, experience and learning in the process of use. If we change our choices easily, it’s hard to become an expert.

Originally published on Medium

Data Pipeline: From ETL to EL Plus T

New challenges lead to new tools, then new challenges?

In the data pipeline, we often talk about ETL, aka Extract-Transform-Load, which is actually a very simple process as follows.

Given an application extracts data from a source and saves it to another data storage after some processing and conversion, this is what ETL does. This is a complete ETL process, and to be honest, most non-ETL applications are doing something like that.

Let me use a simple Python example to demonstrate the implementation of ETL.

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

# Open the CSV file containing the data  
with open("orders.csv", "r") as file:  
  # Use the CSV module to read the data  
  reader = csv.DictReader(file)  
  
  # Iterate over the rows in the CSV file  
  for row in reader:  
    # Extract the data from the row  
    order_id = row["order_id"]  
    customer_name = row["customer_name"]  
    order_date = row["order_date"]  
    order_total = row["order_total"]  
    # Insert the data into the database  
    cursor = db.cursor()  
    cursor.execute(  
      "INSERT INTO orders (order_id, customer_name, order_date, order_total) VALUES (%s, %s, %s, %s)",  
      (order_id, customer_name, order_date, order_total)  
    )  
    db.commit()

This is a simple ETL example, we extract each line from a CSV file (data source), and convert the order details from CSV format to the corresponding columns in the database, and finally write to the database.

As the database evolves, this process also evolves, and the ETL also has a variation, ELT, which seems to be a sequential exchange of T and L. But behind the scenes, the database has actually become more powerful.

Let’s continue with the above example, and see how would ELT be handled?

1
2
3
4
5
6
7

with open("orders.csv", "r") as file:  
  # Use the LOAD DATA INFILE statement to load the data into the database  
  cursor = db.cursor()  
  cursor.execute(  
    "LOAD DATA INFILE 'orders.csv' INTO TABLE orders FIELDS TERMINATED BY ',' ENCLOSED BY '\"' LINES TERMINATED BY '\n' IGNORE 1 ROWS"  
  )  
  db.commit()

In this example, we use MySQL’s LOAD DATA command to write a CSV file directly into the database, converting it during the writing process, rather than converting the format first and then writing it to the database.

In the data pipeline, it is actually various combinations of ETL and ELT, and eventually we will produce a data structure to fit the analysis and utilization.

This process is the same concept in the big data domain, except that the data pipeline of big data multiplies a much larger volume of data and needs to face a more diverse data storage.

From ETL to EL plus T

In my previous articles, we have already discussed the implementation of many big data pipeline architectures.

• Evolutionary Data Infrastructure
• The Infrastructure Stack for Real-Time Data Analysis

For the data source part, we rely heavily on Debezium to capture data changes (CDC).

The process underlying Debezium is that a group of workers extracts the binlogs of various databases, restores the binlogs to the original data, and identifies the contents of the schema to be passed on to the next data storage.

In other words, each data source must have a group of Debezium workers to be responsible for it.

When the number of data sources is large, it is a huge effort just to build and maintain the corresponding Debezium. After all, the software does not work properly when left alone, but also requires a complete monitoring and alerting mechanism, which will eventually create a huge operation overhead.

In fact, E and L in ETL are the most worthless things in a pipeline. For a pipeline to be effective, the most important thing is to produce effective data based on business logic, and T should be the part we should be focusing on.

However, we have to take a lot of time on E and L in the operation of the pipeline, which is not what we expect.

So, is there a solution to this dilemma?

Yes, absolutely, Airbyte, and there are many similar solutions, but Airbyte is open source and relatively easy to use.

Let’s take MongoDB to Kafka as an example, as soon as the setup is done it will be like the example below.

Airbyte can sync automatically or manually according to the settings, and it can choose incremental sync or full sync. Each sync record is listed, and the sync process can be seen after expanding logs.

As a result, the creation of data pipelines is effectively simplified, and data storage can be integrated smoothly through Airbyte’s Web UI.

Conclusion

This article is not to promote Airbyte, nor is it to introduce the Airbyte architecture in depth, but from the standpoint of an architect, we will find that in the domain of big data, we will face various challenges, and each challenge has a corresponding solution, but also requires high learning costs.

In fact, Airbyte also has its limitations, for example, the synchronization period must specify a regular time interval, such as every hour, even though it can use cron representation, but still limited by the fixed time, can not do the real streaming, i.e., once the data is triggered.

In addition, when relying on additional tools, it can be frustrating to encounter this problem.

Once the problem occurs, where to start to troubleshoot, and how to clarify the scope of the disaster, all need to experience and in-depth learning before there is a way to deal with.

Learning these tools is also the biggest challenge for data engineers and data architects, there are many tools to solve many problems. However, big data is really big, and learning all tools is almost impossible.

Finally, let me conclude with a picture.

What’s the next?

Originally published on Medium

The difference between the perspective of a backend engineer and a data engineer

Although I have been working with data in the past, this year, I have been working closely with data engineers and finally fully realized the concept of data-centric.

We all know that applications are made up of data, so as a backend engineer or architect, we must be involved in all kinds of data storage and data structures. How to get the most efficiency out of all kinds of data storage is the goal of every backend engineer, and our focus is on maximizing the throughput of data storage with the most efficient data structures.

There are many technical details involved, including but not limited to the following.

1. Indexing: improve database query performance.
2. Sharding: reduce data volume of datasets.
3. Caching: use faster medium.
4. Denormalization: pre-aggregate complex query results.

But for data engineers, the focus is completely different.

How to find the answer to a question quickly with a lot of data at hand?

Quick here has a very different meaning to what a backend engineer cares about. Data engineers care about quickness in terms of productivity, so they can use all kinds of complex SQL queries, let’s say over 100 lines, to find the answer, even though the SQL query takes a long time.

This is the opposite of what the backend engineers care about, which is the speed of response. From the backend engineer’s point of view, such a complex and time-consuming query is unimaginable.

Data engineers have different data requirements in order to achieve their goals.

1. ETL and ELT: make data more structured.
2. Data Lakehouse: continuous data cleanup and compilation.
3. Governance: manage data catalogs and lineages.

How to organize the data from many sources into a structure that is suitable for use, and put it in an easy-to-access place and reduce the complexity of maintenance is the focus of data engineers.

Therefore, let me summarize this year’s experience in one sentence.

Only those who focus on data and care about data are data engineers.

This is the meaning behind data-centric, and I have spent a lot of time this year learning and changing my mindset in order to understand this.

Data-centric Monitoring

In fact, the concerns of backend engineers and data engineers are also reflected in the metrics of the monitoring system.

The application quality metrics usually include the following key points.

1. response latency
2. throughput
3. error rate

However, for data applications, the quality metrics will be as follows.

1. Freshness: time lag between source and landing.
2. Completeness: whether data is lost at each stage of transformation.
3. Correctness: similar to the above, whether each stage of transformation is correct.

Although, both are applications, they are very different for telemetry. Of course, the underlying infrastructure is also different.

Data Team Organizations

Let’s break down the composition of a data team.

From my point of view, a complete data team would have at least two roles with completely different skill sets and totally different responsibilities: data engineer and data analyst.

In my opinion, there is a simple dichotomy between how to differentiate these two roles.

• Data Analyst: the person who uses data to answer questions.
• Data Engineer: the person who generates the “data” above.

Of course, there are two other roles that can be created depending on how the questions are answered.

• Data Scientist: the person who answers the question through himself.
• Machine learning engineer: answers questions through AI.

But for a healthy data team, I believe there needs to be at least a distinction between the roles of data engineer and data analyst. Just as the focus of a backend engineer is different from that of a data engineer, the focus of a data engineer is different from that of a data analyst.

Conclusion

Throughout 2022, I learned a lot of new experiences and was able to find my blind spots in my domain knowledge in a new role.

As an architect, I believe I still have a lot to learn, so please continue to share and discuss with me in the new year.

Originally published on Medium

Playing Window Function in Postgres

Demonstrating some common practices of SQL Window Function

When doing ETL, we always write data into the database one batch at a time, which contains the snapshot of each batch. these snapshots are basically historical data and may contain a lot of duplicates, so how do we find the data we need in these batches?

This article will use a simple scenario to describe some of our common processing methods, mainly window function.

User Scenario

Suppose a company uses a BPM system to manage employee changes and salary changes, then a simple BPM system will usually have two tables with the details of the employee and the salary of each person as follows.

When working with relational databases, we usually normalize the data, i.e., we plan the tables in terms of requirement dimensions and keep only the necessary fields, which are related to each other by foreign keys.

But in the data analysis context, we often need a fact table, which means everything is aggregated together. An example of a fact table for salary analysis might look like the following.

The common practice in aggregation is batch processing ETL, where each time the required columns are fetched from the employee details table and salary table and saved for later analysis such as salary increase trend. The fact table after the ETL aggregation is as follows.

This is an ETL for batch processing in days, and the two tables will be aggregated every day, so there will be job_process_id for batch processing and create_time for data creation.

Here is the SQL used in the example.

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

CREATE TABLE "public"."cdc_test" (  
	"id" serial NOT NULL,  
	"create_time" DATE NOT NULL,  
	"job_process_id" INT NOT NULL,  
	"empid" INT NOT NULL,  
	"ename" VARCHAR(100) NOT NULL,  
	"sal" INT NOT NULL,  
	PRIMARY KEY ("id")  
);  
  
INSERT INTO "public"."cdc_test" ("id", "create_time", "job_process_id", "empid", "ename", "sal") VALUES  
(1, '2018-12-10', 111, 1, 'Ravi', 3000),  
(2, '2018-12-10', 111, 2, 'Raj', 4000),  
(3, '2018-12-10', 111, 3, 'Ram', 5000),  
(4, '2018-12-11', 112, 1, 'Ravi', 3000),  
(5, '2018-12-11', 112, 2, 'Raj', 4000),  
(6, '2018-12-11', 112, 3, 'Ram', 5000),  
(7, '2018-12-11', 112, 4, 'Srini', 6500),  
(8, '2018-12-12', 113, 1, 'Ravi', 7000),  
(9, '2018-12-12', 113, 2, 'Raj', 4000),  
(10, '2018-12-12', 113, 3, 'Ram', 5000),  
(11, '2018-12-12', 113, 4, 'Srini', 6500);

In this example there are two state changes.

1. Srini was on board on 2018-12-11.
2. Ravi got a pay raise on 2018-12-12.

Get the latest state

As we can see from the above figure, the fact table generated by ETL will contain a lot of duplicate data, if we need to know the current state and to exclude duplicates, how to do? This is where the window function comes into play.

1
2
3
4
5
6
7
8
9
10
11

SELECT  
  t.*  
FROM (  
  SELECT  
    cdc_test.*,  
    (ROW_NUMBER() OVER (PARTITION BY empid,  
      ename ORDER BY create_time DESC)) AS seqnum  
  FROM  
    cdc_test) t  
WHERE  
  t.seqnum = 1;

By using PARTITION BY to lock the fixed columns (empid, ename), and sorting the time in reverse order, we can know the first one will be the latest state.

The result is as follows.

Get changes per salary

With the latest state, we usually still want to know about every salary change. We can still achieve our goal through the window function.

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

SELECT  
  t.*  
FROM (  
  SELECT  
    cdc_test.*,  
    (ROW_NUMBER() OVER (PARTITION BY empid,  
      ename,  
      sal ORDER BY create_time)) AS seqnum  
  FROM  
    cdc_test) t  
WHERE  
  t.seqnum = 1

This time, we PARTITION BY to lock the target is (empid, ename, sal), because we want to treat the same three columns as the same group, then once the sal changes will create a new group.

From the results, we can see that Ravi got one raise, from 3000 to 7000.

Get changes per salary and include the result before salary increase

In the above two examples, we only use ROW_NUMBER(), but in fact there are many other functions available in the window function.

This time, we not only want to get the salary change, but also want the query result to include the column before the salary increase, so we can use LAG().

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

SELECT  
  t.*  
FROM (  
  SELECT  
    cdc_test.*,  
    (LAG(sal) OVER (PARTITION BY empid,  
      ename ORDER BY create_time)) AS prev_sal  
  FROM  
    cdc_test) t  
WHERE  
  prev_sal IS NULL  
  OR prev_sal <> sal;

Similar to the above, except that another window function is used and the WHERE clause is changed.

Now we can see a new column.

Reference

Originally published on Medium

Three Pagination Approaches with SQL and ElasticSearch Demonstration

Pagination is a common technique for web page presentation. When there is a lot of returned data, either to reduce the load on the backend or to improve the user experience, it is usually a good idea to limit the amount of presentation and keep the option to continue browsing.

Here’s a quick look at the most common appearance.

< [1] | 2 | 3 | 4 | ... >

The user can know the current page number, choose the previous or next page, or even jump directly to the specified page. There are three types of pagination approaches as follows.

• Offset Pagination
• Keyset Pagination
• Cursor-based Pagination

But this article is to introduce ElasticSearch pagination, so I will briefly describe these three.

Offset Pagination

This is the most common pagination pattern, and let’s represent it in SQL.

1
2
3

SELECT *  
FROM table_name  
LIMIT 10 OFFSET 20;

Suppose there are 10 records on one page, then we can get the result of the third page by this command. Use LIMIT to limit the number of records per page and use OFFSET to jump to the specified page.

There are many advantages of such a pagination.

1. It is very easy to implement, as well as very intuitive.
2. It is possible to jump to a random page.

But the disadvantages are also obvious.

1. Performance is a problem.

Why is there a performance problem?

In the above example, it looks like we only need to take 10 records from the database, but in fact, the database will take 30 records and discard the first 20 records. When OFFSET is very large then the database will still take all the records and create a lot of overhead.

Keyset Pagination

In order to solve the performance problem, we have the keyset pagination approach.

We still use SQL as an example, following the above scenario, we limit the number of records per page to 10 and take the third page.

1
2
3
4
5

SELECT *  
FROM table_name  
WHERE id > 20  
ORDER BY id  
LIMIT 10;

Using LIMIT to limit the number of records per page is the same as above, except that we first sort by ORDER BY and then set the starting position by WHERE instead of OFFSET. In this way, we can get the third page without taking 30 records.

There are several implementation details that must be paid attention.

1. Must be able to sort quickly, in the case of relational databases, the columns to be sorted must have indexes.
2. Where do the WHERE clause come from? It can be from the frontend with the last position of pages, or it can be the backend through some storage mechanism.

Of course, there are advantages and disadvantages to such a pagination. The advantages are as follows.

1. It works well when implemented correctly, e.g., on index columns.
2. It can run on very large data sets.

But these advantages come at a price.

1. difficult to implement
2. no way to jump to random pages

I believe the first drawback is easy to understand, after all, it is not a very intuitive approach, the WHERE clause is determined by engineering methods.

The second drawback is to jump to a specific page, there must be a correct WHERE clause, but this WHERE clause depends on the last page jump result, whether it is frontend or backend processing. Therefore, users cannot jump pages as they wish.

Cursor-based Pagination

This is an advanced version of keyset pagination, which is actually a special case of cursor-based pagination.

A cursor is an object defined by the engineering side to mark where the pagination starts. There are several common types of cursors.

1. Encoded cursor, such as base64. Suppose the cursor is eyJpZCI6IDIwfQ==, then we will find out it is a JSON format string and the specified id is 20.
2. Token cursor, the backend generates a token for each search result and stores the token, the frontend can use the token to specify the jump page, the backend can know where to start from based on the token.

No matter which cursor is used, the backend still uses the keyset pagination mechanism for pagination.

The advantages and disadvantages of cursor-based pagination are fully inherited from keyset pagination, but cursor has the extra advantage that cursor can store more information, such as session timeout, user privilege, etc.

How about ElasticSearch

The examples above use SQL as an example, but in fact ElasticSearch supports these methods as well.

Here is an example of offset pagination.

1
2
3
4
5
6
7
8

GET /index_name/_search  
{  
  "from": 20,  
  "size": 10,  
  "query": {  
    "match_all": {}  
  }  
}

Similar to the principle of SQL, from is used to specify where to start, and size is used to limit the number of records per page. Then, of course, there are also scalability limitations, and moreover, ElasticSearch directly limits the maximum amount of from and size in order to avoid poorly performing queries from affecting the health of the cluster.

• max_result_window: The maximum amount of data that can be searched at once, the default is 10000. If from + size > 10000, it will directly return an error.

How to solve it? Use the keyword search_after for keyset pagination.

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

GET index_name/_search  
{  
  "size": 10,  
  "sort": [  
    {  
      "id": {  
        "order": "asc"  
      }  
    },  
  ],  
  "query": {  
    "match_all": {}  
  },  
  "search_after": [  
    20  
  ]  
}

This is a typical implementation of keyset pagination on ElasticSearch.

So does ElasticSearch have support for cursor-based pagination? Yes, and unlike SQL, you have to implement your own cursor in the application. ElasticSearch already has its own cursor mechanism, called the Scroll API.

The principle is to create a snapshot of the current query and jump to the next page by calling scroll every time. A practical example would look like the following one.

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

POST index_name/_search?scroll=1m  
{  
  "size": 10,  
  "sort": [  
    {  
      "id": {  
        "order": "asc"  
      }  
    },  
  ],  
  "query": {  
    "match_all": {}  
  }  
}

This will get a response with _scroll_id, and then we can jump to the next page just by using this _scroll_id.

1
2
3
4
5

POST /_search/scroll  
{  
  "scroll": "1m",  
  "scroll_id": "OXOXQQ=="  
}

Since it is a cursor-based pagination, we cannot specify which page we want to jump to, we can only keep going to the next page until there is no result.

Because Scroll API is a snapshot of the current query result, so we must carefully choose the TTL or delete the snapshot after finishing, otherwise it will occupy the hard drive space.

1
2
3
4

DELETE /_search/scroll  
{  
  "scroll_id" : "OXOXQQ=="  
}

In the new version of ElasticSearch, it is no longer recommended to use Scroll API for deep pagination, instead, another new mechanism (released after 7.10), PIT (Point In Time).

PIT works similarly to Scroll API, but is more flexible and better optimized for performance.

Scroll API takes a snapshot of a single query and can only jump pages on that snapshot, but PIT takes a snapshot of the current data set and can do anything after getting the snapshot, not just jump pages.

However, this is not related to the pagination, so I will not dive into PIT in this article.

Conclusion

In general, there are two major types of pagination, offset pagination and keyset pagination, and only keyset pagination can run on big data, but if you want to jump pages randomly, only offset pagination can, which is a trade-off.

In fact, in a big data scenario, if the feature requirements strictly define the maximum number of pages, then even using offset pagination will not affect performance.

Therefore, it is not just a technical decision to choose the implementation method, but more often a trade-off in terms of feature requirements. Nevertheless, engineers should know if there is a requirement for random page jumping on big data without page limit, it is time to say no, please clearly reject.

Originally published on Medium

How to Determine API Slow Downs

A statistical solution to know if its time to refactor your code

API slowdown has a great impact on the user experience or even loss of trust, so we always want to know the latency as early as possible.

Therefore, it is common to integrate performance tests and measurement benchmarks into automation testing.

But the concept of slow is very difficult to evaluate.

• Did it happen by accident? Or is it an ongoing status?
• Is it caused by the experimental environment? Or is it a side effect of the code?
• Is it the result of increased features? Or is it a defect?

It is difficult to have a clear answer to these questions.

Assuming we have a way to detect “slow”, then we will send an alarm or a report. But if it’s just a “coincidence” or a false alarm caused by experimental environments, eventually people will try to ignore these signals.

In addition, it is to be expected that APIs will slow down as features are added. When we have more parameters, more business logic, and more if-else, it’s no wonder that the response time slows down.

Furthermore, if we use threshold to avoid the above problem, then when we really recognize API slowdown, we won’t know what the root cause of the slowdown is.

All these factors make the task of detecting API slowdowns extremely hard.

It’s time for statistics to come into play

To address the mentioned issues, let’s summarize the requirements.

1. to be able to know specifically when the problem occurs
2. to be able to allow accidental outliers
3. to be able to avoid the normal phenomenon of slow rise
4. to be able to issue alarms

First, let’s look at a diagram.

• The horizontal axis indicates each test, which can be considered as time or release version, or even if integrated into CI/CD pipeline, that is each commit.
• The vertical axis is the latency of the target tests.
• The red line is the main character of this article, Linear Regression.

From the diagram, we can know before a certain time, basically the latency is slowly increasing, but after a certain time, the whole latency is rapidly increasing. And that time point is the problem we want to know.

Perhaps you will say, it is easy to find out by the naked eye by looking at the numbers each time? After all, it is quite intuitive.

The key is the scale of the latency. If the average of the data points from the group below is 0.1 seconds, and the group behind is 0.2 seconds, will you still feel very intuitive?

Therefore, we need a brain-friendly standard, i.e., linear regression.

Through linear regression, we can know the overall trend of the data set is up. However, this is not enough, because just knowing that the trend is up does not mean we can judge whether it is normal or abnormal.

The group of dense data points in the front is also up, so how can we determine that the trend is abnormal and the front is normal?

Let’s look at another diagram.

Question: Is the latency of this diagram normal or abnormal?

It looks very similar to the previous diagram, there is a clear upward trend, so it should be abnormal, right? Really?

Let’s add some reference points and zoom out a bit.

Have you found? The diagram above is actually a normal part of the original diagram, just a slow rise.

By simply doing linear regression analysis on a single data set, we may get wrong conclusions because we have no benchmark for comparison and cannot tell whether it is a natural phenomenon caused by increased features or a problem caused by defects.

Solution

Back to our question, how can we detect the problem as early as possible without false positives?

First, we need to choose a reference point, e.g., one day ago. Then, we perform a linear regression on the entire data to get a red line. Of course, we also need to perform a linear regression on the data before the reference point, which will lead to the green line.

The slope of the red line and the green line is used to calculate the angle as the judgment basis. In other words, when the angle between the red line and the green line is larger, then it is more likely to be abnormal.

Therefore, the threshold for setting an alarm is neither the latency itself nor the slope, but the angle of the two lines.

Given the slopes of the two lines are k1 and k2, then the formula for the angle θ is as follows.

In Python, it would be:

1
2
3
4
5

import math  
thera = math.degrees(math.atan2(abs(k2 - k1), abs(1 + k1 * k2)))  
# atan2 does not need to handle denominator zero error cases  
# If we use atan  
# we need to solve the case where k1 * k2 is equal to -1 first

We can do several experiments with simulated values to get our ideal alarm threshold.

Finally, in addition to the angle threshold, we also need another alarm indicator, which is the upper bound of the allowed latency.

Why do we need to set an upper bound?

Because if our latency keeps increasing in a green line trend, one day it will reach an unacceptable value, but if we just set the angle threshold, we will never know how bad the situation is until we have a disaster on the production.

Finally, the following is an example of how to compute linear regression in Python.

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

import numpy as np  
from sklearn.linear_model import LinearRegression  
  
# raw_x and raw_y are the results of each test  
# raw_x can be a time, version or commit normalization  
# raw_y are latencies  
x = np.array(raw_x)[:, np.newaxis]  
y = np.array(raw_y)[:, np.newaxis]  
  
bfl = LinearRegression()  
bfl.fit(x,y)  
y_pred = bfl.predict(x)  
  
slope = bfl.coef_[0, 0]

Conclusion

So, how do we know if the API is slowing down? We need two alerts.

1. the angle between the current data line and the data line of the reference point after linear regression.
2. the maximum upper bound of the latency.

When we have these two alarms, we can know exactly when the code is causing problems and if it is time to refactor the system.

In fact, this is a statistical approach to achieve the goal, but there are still some drawbacks to this. For example, how to set the threshold value of the slope angle? How to choose the reference point? What is the statistical range to be chosen? All these questions require experiments to find the answers.

Therefore, the most effective way is to introduce machine learning into the system of latency analysis, and judge whether the current result is normal or abnormal through the prediction model, rather than through human settings. It’s more difficult for humans to make judgments than machines.

However, to build a machine learning model requires corresponding domain knowledge and skills, which may not be suitable for small organizations, so linear regression is a reliable and low-cost solution.

Originally published on Medium

The Infrastructure Stack for Real-Time Data Analysis

A look at the popular options available

Previously, we introduced the evolution of data infrastructure, where we evolved the original data monolith step by step into an architecture that can support both real time analysis and various data governance.

However, in that article, we didn’t cover what technical selections were available but only a high-level view of the architecture evolution process.

In this article, we will focus on the stack of real-time analysis and list some hot or gradually popular options.

Real-Time Infra

Before starting, let’s briefly introduce the whole architecture of real-time analysis.

As we mentioned before, the core of the entire real-time infrastructure is streaming.

In order to quickly process all events from event producers, streaming plays an important role, and within streaming, there are stream platforms and stream processors. Stream platform is the broker of streaming, which is used to store and distribute streams to processors. And the processor will send the stream back to the platform after processing the stream.

Why don’t processors deliver the streams directly backwards?

Because the stream may go through more than one processor. To complete a full use case, the stream may go through many processors, each of them focusing on what they are intended to do. This is the same concept as a data pipeline.

When the streaming is processed, it is persisted and is available for users as needed. Therefore, the serving layer must be a data store with high throughput and provide a variety of complex queries. For this requirement, traditional RDBMS cannot meet the throughput requirements, so the serving layer is usually not a relational database.

The final front is for presenting the data to the end-user, which can be tables, diagrams, or even a complete report.

Event Producer

We already know that event producer are responsible for generating various “events”, but what kinds of events exactly are there?

There are three types.

• Existing OLTP database
• Event tracker
• Language SDK

Existing OLTP database

Any system will have a database, whether it is a relational database or a NoSQL database, and as long as the application has storage requirements, it will use the database that suits its needs.

To capture data changes from these databases and deliver them to the streaming platform, we often use Debezium.

Event tracker

When a user operates a system, whether it is a web frontend or a mobile application, we always want to capture those events for subsequent analysis of user behavior.

We want the trackers to be able to digest the rapidly generated events for us, but we also want them to have some customization, such as enriching the events. Therefore, plug-ins and customization will be a priority when selecting a tracker.

Common options are listed below.

• TWILIO segment
• SNOWPLOW

SNOWPLOW provides an open source version, while segment does not.

Language SDK

The last is the events generated by various application backends, which are delivered through the SDK provided by the stream platform. The technical selection here will depend on the stream platform and the programming language in use.

Stream Platform

The concept of stream platform is very simple, it is a broker with high throughput.

The most common option is Kafka, but there are also various open-source software and managed services. By the way, the following order does not represent the recommendation order.

Open Source

• Redpanda
• Apache Pulsar

Managed Service

• AWS Kinesis
• GCP Pub/Sub
• Azure Event Hub
• Confluent
• upstash
• ably
• solace

Stream Processor

A stream processor, as the name implies, is a role that handles streams. It must have scalability, availability, and fault tolerance, and the ability to support various data sources and sinks is also an important consideration.

The Apache Flink, which is often mentioned, is one of these options, and there are many others.

Open Source

• Apache Kafka Streams
• ksqlDB
• Apache Beam: Streaming framework which can be run on several runner such as Apache Flink and GCP Dataflow
• Materialize: Streaming database like ksqlDB and based on Postgres
• MEM GRAPH: Graph database built for real-time streaming and compatible with Neo4j
• bytewax: Python framework
• Faust: Python framework

Managed Service

• GCP Dataflow
• DELTA STREAM
• Quix

Serving Layer

The function of the serving layer is to persist the results of the stream processing and make them easily available to users. Therefore, it must have two important conditions, first, a large throughput, and second, the ability to support more complex query operations.

In general, there are two different approaches, one is to choose a common NoSQL database, such as MongoDB, ElasticSearch or Apache Cassandra. All of these NoSQL databases have good scalability and can support complex queries. In addition, these databases are very mature, so the learning curve is low for both use and operation.

In addition, there are also some NoSQL databases rising up for low latency and big data, e.g., SCYLLA.

On the other hand, there are many newcomers to the SQL family. These new SQL-compatible databases have a completely different implementation logic than traditional relational databases, and thus have a high throughput and can even interact directly with stream platforms.

Moreover, these databases have the scalability that traditional RDBMSs do not have, and can still have low query latency in big data scenarios.

• Apache Pinot
• Apache Druid
• ClickHouse
• ROCKSET

Frontend

The most common in the frontend is still to build services using common web frameworks, such as these three most commonly frameworks.

• AngularJS
• Vue.js
• React

Furthermore, low-code frameworks are becoming popular in recent years. Low code means developers only need to write a small amount of code to use plenty of pre-defined functions, which can significantly shorten the development time and speed up the release.

In order to make the data more real-time, the purpose is to make the value of the data as early as possible, so agile development in the production environment is also a reasonable consideration. Here are two popular low-code frameworks.

• Streamlit
• Dash

Finally, there are various data visualization platforms.

• Apache Superset
• Metabase
• redash
• PowerBI

Conclusion

Although this article lists a lot of targets for technical selection, there are definitely others that I haven’t listed, which may be either outdated, less-used options such as Apache Storm or out of my radar from the beginning, like Java ecosystem.

Besides, I did not put links to the three major public cloud platforms that are already relatively mature (AWS, GCP, Azure), because those can be found in many resources on the Internet at any time.

Even though these technical stacks are listed by category, some fields actually overlap. For example, although Materialize is classified as a stream processor, it makes sense to treat as a serving layer because it is essentially a streaming database, and the same is true for ksqlDB.

Each project has its own strengths and applications. When making a choice, it is important to consider the existing practices and stacks within the organization, as well as the objectives that you want to accomplish, so that you can find the right answer among the many options.

If there is a project I didn’t list and you feel it is worth mentioning, please feel free to leave me a comment and I will find time to survey it.

Originally published on Medium

Insights from Worldwide Software Architecture Summit vol.2

Six months ago, I attended Worldwide Software Architecture Summit (WSAS’22), where many sessions were discussing complexity, especially the complexity of microservices. At that time, I wrote two insights as follows.

• Original Sin of Microservices, Part 1
• Original Sin of Microservices, Part 2

The main point is building a microservice architecture is not always the best solution, the complexity of microservice architecture is very high and there are many trade-offs, how to find a relatively good option among many choices is a challenge.

Recently, WSAS’22 was held for the second half of the year, and I also attended.

Among those sessions, Conway’s Law was mentioned several times.

In order to maintain the microservice architecture, the organization structure also needs to be adjusted to properly maintain many services. In the reorganization, how to let engineers do their work more efficiently also became an important issue.

During the three-day session, I came to one conclusion.

The productivity of engineers can only be increased by focusing on the happiness of engineers.

How to do that? Well, it’s not that hard.

1. Optimize the onboarding process. Let new engineers get into work mode faster.
2. Be able to local development. Although the software architecture is moving towards microservices, engineers need to be able to build a development environment with the equipment at hand.
3. Establish software testing, whether it is unit testing, integration testing or even continuous integration, having a good testing framework to find defects as early as possible will make people more motivated.
4. Reduce the cognitive load. This covers a wide range of topics, including complete documentation to guide engineers, service templates to reduce the initial process of developing a new service, and a good continuous deployment process to reduce the background knowledge on operations.
5. Have the permission to do whatever needs to be done, otherwise it will be stuck in various approval processes.

These five items are more from the engineer’s point of view and I believe they are easy to understand.

If you can focus on what needs to be done and have a good way to get it done, then the engineer should be happy.

But there is another item I feel is the most important.

Improving the code quality.

I believe we can all agree poor code can have a serious impact on productivity, but how much? Is there any way to measure it?

One session is a static scanning company, they published a paper, which mentioned that

1. 15 times more bugs in poor code than good code.
2. the development speed of poor code is 2 times slower than good code.
3. the feature delivery time of poor code is 9 times longer than that of good code.

From these supporting evidence, it is also indirectly proven that if an engineer has to suffer from poor code, he will not be happy.

From my point of view, microservice governance is much more important than microservice architecture. As an architect, I put more emphasis on how to practice than the perfect architecture. An architecture that makes engineers happy is a good architecture, whether it’s a monolith or a microservice.

Originally published on Medium

Solving Dogpile Effect Using Python

A workable code solution

Last time, we talked about how to solve Dogpile Effect.

Three approaches are mentioned as follows:

1. Warm Up Cache
2. Extend Cache Time
3. Exclusive Lock

However, we also mentioned that each of the three approaches has its own applicable scenario and its corresponding potential risks. So is there a way to extract the advantages of each and construct a more complete solution?

This article will provide an example and explain my ideas.

Solution Concept

Extending the cache time effectively improves the availability of the cache. When the cache is not valid and is requested at the same time, only one request can get through the cache and into the backend system. The rest of the requests will get the original result as the cache time is extended.

However, when concurrent requests occur at the same time ( which is generally uncommon), there is still the possibility of multiple requests entering the backend system, hence the exclusive lock approach.

Nevertheless, the cost of using exclusive locks all the time is too high, and we should try to minimize the use of exclusive locks if possible. Then, use the exclusive lock only when the cache does not exist and there is a need to access the backend system, else just use the extend cache time.

The whole process is as follows.

First of all, determine whether the cache exists, if the cache exists we still have to determine whether the cache is expired. If everything is fine, we can just take the original value of the cache, but if the cache is expired, we must enter the process of updating the cache.

In order to avoid the impact of high concurrent requests, all update cache processes should try to acquire a lock.

On the other hand, if the cache does not exist from the beginning, then the process of updating the cache will be the same. Only the process is different as mentioned above, because there is no original value, so those who did not acquire the lock must wait for the lock before they can get the result.

Solution Overview

Before we get into the details of the implementation, let’s look at the actual practice.

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

def read_aside_cached(ttl, lock_period, race_period):  
  def decorator(func):  
    def wrap(*args, **kw):  
      key = f"{func.__name__}_{args}_{kw}"  
      return cache_factory(key, ttl, lock_period, race_period).handle(func, *args, **kw)  
        
    return wrap  
  return decorator  
  
@read_aside_cached(60 * 5, 30, 60)  
def foo(a, b=1, c=2):  
  return db.query(a, b, c)

This is an example in Python where we use a decorator to encapsulate an actual database operation.

This decorator requires several parameters.

1. ttl, this is easy to understand, it is the expiry time of this cache.
2. lock_period, because we need to acquire the lock, so this parameter determines how long we have to lock.
3. race_period, this parameter is used to determine how long we want to extend the cache.

In the above example, foo has a cache expiry time of 5 minutes and retains a 1 minute buffer. The lock time is 30 seconds, which is related to the expected time of the database operation.

Solution Details

Next, let’s break down the actual details of the flowchart.

1
2
3
4
5
6
7
8
9
10

def cache_factory(key, ttl, lock_period, race_period):  
  value, expired_at = Store.get(key)  
    
  if expired_at is not None:  
    handler = ExistedCacheHandler(key, ttl, lock_period, race_period)  
  else:  
    handler = NonExistedCacheHandler(key, ttl, lock_period, race_period)  
  
  handler.set_meta(value, expired_at)  
  return handler

At the beginning of the flowchart, we need to try to get a cache first and use the results to see if we need to extend the cache time.

The top and bottom paths of the flowchart are encapsulated by each class. Let’s look at the implementation of ExistedCacheHandler first.

1
2
3
4
5
6
7
8
9

class ExistedCacheHandler(BaseCacheHandler):  
  def handle(self, func, *args, **kw):  
    if self.now > self.expired_at and Store.try_lock(self.key, self.lock_period):  
      result = func(*args, **kw)  
      Store.set(self.key, result, self.ttl + self.race_period)  
      Store.unlock(self.key)  
      return result  
  
    return self.orig_val

If a cache has expired and successfully acquires a lock, it is responsible for updating the cache.

In the previous article, we introduced the Rails approach, where Rails writes the original value back to the cache again and extends the valid time slightly. But here, we directly let the cache time be (ttl + race_period), so we don’t need to extend the cache time manually.

On the contrary, if the cache has not expired or has not been locked, then the original result in the cache is used.

On the other hand, the logic of cache not exist is more complicated.

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

class NonExistedCacheHandler(BaseCacheHandler):  
  def handle(self, func, *args, **kw):  
    while self.expired_at is None:  
      if Store.try_lock(self.key, self.lock_period):  
        result = func(*args, **kw)  
        Store.set(self.key, result, self.ttl + self.race_period)  
        Store.unlock(self.key)  
        return result  
      else:  
        while not Store.try_lock(self.key, self.lock_period):  
          time.sleep(0.01)  
          self.orig_val, self.expired_at = Store.get(self.key)  
          
        Store.unlock(self.key)  
    else:  
      return self.orig_val

When a cache is found to be non-existed, we still have to acquire the lock in order to update the cache. But if the lock is not acquired successfully, we must wait, either for the lock or for the cache to be updated.

Why should we wait for either of these two conditions?

The reason is the person who acquired the lock may not have released the lock for “some reason”. Our ultimate goal is to get the cache result, so even if we don’t get the lock, we still get the result. Of course, if the lock is successfully acquired, the responsibility of updating the cache will be taken up.

Finally, let’s look at two common components.

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
28
29

class Store:  
  @staticmethod  
  def get(k):  
    value = redis.get(k)  
    expired_at = redis.pttl(k) / 1000 + time.time() if value is not None else None  
    return value, expired_at  
      
  @staticmethod  
  def set(k, v, ttl):  
    return redis.set(k, v, "EX", ttl)  
  
  @staticmethod  
  def try_lock(k, lock_period):  
    r = redis.set(k, 1, "NX", "EX", lock_period)  
    return r == "OK"  
  @staticmethod  
  def unlock(k):  
    redis.del(k)  
      
class BaseCacheHandler:  
  def __init__(self, key, ttl, lock_period, race_period):  
    self.key = key  
    self.ttl = ttl  
    self.lock_period = lock_period  
    self.race_period = race_period  
    
  def set_meta(self, value, expired_at):  
    self.orig_val = value  
    self.expired_at = expired_at

The BaseCacheHandler defines the constructors and a helper function.

Store is the core of the whole implementation, and I use Redis as a demonstration.

• get(): In addition to getting the cache value, we also need to get the expiry time of the cache.
• set(): Write the value and also set the expiry time.
• try_lock(): Use Redis’ atomic update to lock with NX.
• unlock(): simply removes the key.

By assembling all these pieces, the cache decorator is complete, not only with the ability to extend the cache time but also with exclusive lock support.

Conclusion

This is a workable example, and we have arranged it in a more intuitive way to make it easier to understand. However, there are some things that could be refined.

For example, many places currently use a single command to operate Redis directly, and it would be better to write it in Redis pipeline. Moreover, it would be a good idea to write some simple logic as a script in Lua.

I have to say such an implementation is actually very complex, but does a read-aside cache really need to do that?

It depends on the load of the application and what we expect from the application.

If the backend system is strong and can handle a sudden spike, then a regular extended cache time can work. But if the backend is weak, it is necessary to consider a more solid approach.

Enhancing the caching mechanism is one option, but enhancing the backend system is also an option. There are several common ways to enhance the availability of the backend system.

1. circuit breaker pattern
2. service degradation
3. multi-layer caching

This article provides an option to enhance the cache, without the need to deploy new components and just modify the logic, in my opinion, it is still worth it.

Originally published on Medium

Solving Dogpile Effect

How to cache in high-volume scenarios

We have discussed cache consistency before, and at that time we mentioned it is possible to achieve good consistency if we implement read-aside cache correctly. To further improve consistency, a more complex solution must be used, e.g., write through cache or write behind cache.

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

Today, we are going to discuss another common scenario of caching, Dogpile Effect.

Dogpile Effect means when the system is under heavy volume of traffic, whenever a cache invalidates, either by cleanup or timeout, it will have a huge impact.

For example, if a cache entry is accessed by 100 requests at the same time, once the entry expires, 100 requests will hit the backend system directly, which is a serious challenge for the backend system.

Therefore, let’s see what approaches are available to deal with Dogpile Effect. There are three common approaches as follows.

1. Warm Up Cache
2. Extend Cache Time
3. Exclusive Lock

Each of these three options has its own pros and cons, and in fact, they are already widely used.

Question, do you know which one is the race_condition_ttl of the cache store in Ruby on Rails*?*

Warm Up Cache

Before we talk about this approach, let’s review the read path of the read-aside cache.

1. First, all read requests are read from the cache.
2. If the cache fails to read.
3. Then read from the database, and write back to the cache.

The problem occurs when the cache is missed.

When there are lots of read requests under a high volume system, it will result in lots of database accesses. If we can keep the cache active anyway, can’t we solve Dogpile Effect?

Thus, the approach is to replace the cache TTL with a background thread that updates all caches periodically, e.g., if the cache TTL is 5 minutes, then update all caches every 5 minutes, so that cache invalidation is no longer encountered.

But such an approach has its drawbacks.

1. It is very space inefficient if applied to all cache entries.
2. In some corner cases, caching still invalidates, such as an eviction triggered by cache itself.

Although this is a simple and brutal approach, I have to say it works very well in some situations. If we have a critical cache that has to deal with a lot of traffic and the cost of updating is high, then keeping that cache fresh is the most effective way.

As long as we avoid updating all cache entries, then it won’t take up a lot of space and will be less likely to trigger corner cases.

Extend Cache Time

Warm up cache can work well on specific critical cache, but if no critical cache is defined at all, then the benefits of warm up cannot be applied effectively.

Therefore, the second approach is suitable for general purposes.

When reading a cache, if a cache timeout is found, then extend the cache time slightly and start updating the cache. If there is a concurrent read request, the later read request will read the cache with the extended time to avoid accessing the backend database simultaneously.

Assuming that the cache TTL is 5 minutes, we set each cache to have a one-minute extension time, as shown in the Gantt chart below.

In the time interval 0–5, reading the cache will get the original value. If someone reads during the time interval 5–6, although the cache expires, the cache will be extended, so the original value will still be available. But at the same time, the first person who reads in the interval 5–6 must be responsible for updating the cache, so after the time point 6, the cache will be updated to the new value.

Let’s use a sequential diagram to represent the two concurrent request scenarios.

Suppose the cache has already timed out. When A fetches, it finds that the cache has timed out, so it first writes the original value back to the cache and performs the regular read operation to fetch the value from the database, and finally writes the new value back to the cache.

However, B finds that the cache has not expired when it reads, because the cache is extended, so it can get the original value.

With this approach, only one of the N concurrent requests needs to access the backend, and the rest of the N - 1 requests can still fetch the value from the cache.

Actually, Ruby on Rails’s race_condition_ttl is this implementation. Lines 855 and 856 in the link above are the operations that extend the cache time.

This approach looks like an effective way to handle high volume traffic, with only one request for access to the backend, right?

The answer is, no, not really.

This is obviously useless when facing a high-concurrency scenario. Let’s continue to describe the problem in a sequential diagram.

The same A and B as earlier, but this time A and B occurred very close to each other. As you can see from the sequential diagram, B has already happened when A tries to write the original value back to the cache, so B also feels he is the first one.

When N requests arrive completely at the same time, then such an implementation still cannot solve Dogpile Effect.

Exclusive lock

Extending the cache time seems to have solved most of the problems, but it’s still not good enough in a high-concurrency system. Therefore, we need a way to serialize the high-concurrency scenario. Previously, we introduced the concept of the exclusive lock.

In this approach, we are trying to avoid multiple concurrent requests to go through the cache by exclusive lock.

First, an exclusive lock must be acquired before the cache can be updated. Only those who can acquire the lock are eligible to update the cache, i. e. access the database, while the rest of the people who do not acquire the lock must wait for one of the following two conditions.

1. whether the cache has been updated
2. whether the lock can be acquired

The wait-and-acquire lock must also verify whether the cache has been updated in order to further avoid duplicate database accesses.

Of course, there is an obvious drawback to this approach. Making the concurrent process serializable would significantly reduce concurrency. In addition, the waiting is an extra overhead that not only consumes resources but also affects performance.

It’s worth mentioning that we often say Redis is not reliable, so in this scenario, is it necessary to use Redlock to further improve reliability?

It depends.

In general, no, even if Redis is not reliable enough to be used exactly once, the rare occurrence of a leak is not a problem in this scenario. This is not a scenario that requires strong consistency, but at best a few accesses to the database.

Conclusion

In order to address a common problem with caching, Dogpile Effect, we have reviewed three common approaches.

The scenario where the warm up cache is applicable is the “critical cache”. Once we can narrow down the scope of the cache, then keeping it fresh is the most intuitive and easy way for us.

Extending the cache time is a general-purpose approach that can effectively tackle a variety of high-volume scenarios. By providing a time buffer, the cache is allowed to serve for a long time until the cache is updated by “someone”.

Exclusive lock is a high-concurrency specialization approach. In order to avoid concurrent requests from hitting the backend system, concurrent requests are turned into sequential requests through a serializable mechanism.

Each of these three approaches has its own advantages and benefits, but in fact, extend cache time and exclusive lock can be combined into a total solution. I will explain the implementation details of this in code in the next article.

Let’s call it a day.

Originally published on Medium