CT Wu

Experience a boost in API response times with our paradigm-shifting move from Semi-Lambda to TiDB & Kappa architecture

We recently launched a big technical revamp, and the results have been impressive. We’ve seen a 9x improvement in response times for specific APIs (from 9 seconds down to 1 second).

You may ask why the API response time is so long. It’s mainly because of the product characteristics, it’s a data product, there will be a lot of data aggregation and analysis, and the API response will be the result of a period of analysis. Therefore, in general, there will be a longer latency.

Nevertheless, through this technical revamp, we have significantly reduced the API latency while maintaining the product features.

This is due to two core changes.

1. Database changes
2. Kappa architecture

Let’s take a closer look at what we did.

Semi-Lambda Architecture

At first, our architecture is a “semi-Lambda” architecture. To understand what a semi-Lambda architecture is, let’s first look at a Lambda architecture.

A typical Lambda architecture has three components.

1. Batch layer
2. Speed layer
3. Serving layer

Data processing from the data source is divided into two paths, one is periodic batch processing, which has the advantage of processing a large amount of data quickly.

On the other hand, in order to enhance the freshness of the analysis, there is real-time processing to calculate the data in the current time period. The results from both sides are aggregated into a serving layer so that end users can get the final results.

For example, if the time is 12:30 and the batch layer runs every full hour, then the data from 12:01 to 12:30 will be processed by the speed layer.

The above is a standard Lambda architecture.

Then, what is a semi-Lambda architecture? The illustration is as follows.

We also have a batch layer and a serving layer, but we are missing the speed layer. Instead, we write the raw data directly into the serving layer, and the client, e.g. API, which is invoked by the upper layer, runs the logic to process the raw data and combine it with the batch result.

In other words, we are the on-the-fly speed layer.

The drawbacks of such an architecture are obvious.

1. Postgres is a monolithic database, so even though it can be read-write split, the replicas still need to synchronize with the master constantly. This is fatal in large raw data write scenarios, where I/O consumes a lot of system resources.
2. Postgres is limited in its capability to handle big data. Although it is possible to make data blocks smaller and easier to read by partitioning, a single database is not enough when the algorithm is designed for a large number of partitions.

Therefore, we need to propose solutions to these two problems.

Streaming ingest problem

First of all, we need to solve the database bottleneck of writing large amounts of data. Therefore, a proper database is essential.

For those who have been following me, you should know I have been studying RTOLAP databases such as Apache Pinot for some time. But in the end, we didn’t go with RTOLAP.

The main reason is RTOLAP database does not perform well in two specific scenarios.

1. Upsert
2. Join

Because of the application features, we need to perform frequent upserts on big data, and in addition, we need to do a large number of joins, doesn’t matter if it’s a cross-table join or self-table join.

On the other hand, it is difficult to say RTOLAP database has comprehensive support for SQL statements, and we want to minimize the amount of trial-and-error work required for migration, so we didn’t choose RTOLAP database in the end.

Fortunately, around the same time, we got some help from a vendor and learned about a distributed database completely compatible with MySQL, i.e. TiDB. Previously, I wrote an article describing TiDB application scenarios. In fact, TiDB can support both OLTP and OLAP scenarios, fitting our needs.

Speed layer problem

Once the database selection is finalized, the next step is to address the lack of a complete speed layer.

There are two possible directions.

1. Build a complete Lambda speed layer.
2. Change to Kappa architecture

In the end, we chose the latter and started to move the architecture to Kappa.

The reason is both Lambda and Kappa need to build streaming processing capabilities, but to build a speed layer for the Lambda architecture we still need to rewrite the batch logic in a streaming framework. It would be better to migrate the batch logic to the Kappa framework so that if there is a need for any feature in the future, we just need to add the new logic to the streaming framework.

This is one of the advantages of Kappa over Lambda, there is no need to maintain two pipelines (batch and real-time) with the same logic.

After moving to the Kappa paradigm, you can see that the whole architecture has become simple, with Flink being responsible for data processing and batch processing disappearing. In addition, Postgres was moved to TiDB, which of course required changes to the query syntax, but it was well worth it.

Finally, we solved the OLAP query latency problem by using streaming pre-processing and the power of the TiDB distributed database.

Conclusion

Paradigm shift is a battle against time, and how to keep optimizing the system while guaranteeing development productivity is a big challenge.

In addition to initiating discussions and a series of technical selections, it is more important to plan the actual migration and develop efficiently. This requires architects, PMs, and developers to work together.

The question is frequently asked: How much of a benefit is this migration? It’s a tough question to answer before getting one’s hands dirty. But if we don’t know the benefits, how can we have the confidence to invest for such a long period of time?

From my point of view, we need to list the problems we want to solve and prioritize them in a way that will affect the final solution. Then, identify one or two showcases to act as pilots, and learn as much as possible from these showcases, rather than launching a large-scale revamp all at once.

When results are achieved in the showcases, we will be able to measure the benefits.

Engineering is different from science in nature. Engineering is learning by doing, not by theorizing. Airplanes fly before scientific theories can explain them, that’s my mindset on technical revamping, let’s see what you guys think too.

Stackademic 🎓

Thank you for reading until the end. Before you go:

• Please consider clapping and following the writer! 👏
• Follow us X | LinkedIn | YouTube | Discord
• Visit our other platforms: In Plain English | CoFeed | Venture | Cubed
• More content at Stackademic.com

Originally published on Medium

Simplify Web App Development: Code Lite, Create Big!

Empower your ideas with Streamlit’s Magic — build feature-rich Web Apps effortlessly

Hello data engineers and backend engineers!

• Have you ever wanted to implement a Web App but didn’t know where to start because you don’t have frontend skills?
• Have you ever struggled to write interactive components and spent a lot of development time?
• Have you ever wanted to present an idea quickly but didn’t have the right tools?

Here’s your savior, let’s welcome Streamlit.

It’s a full-stack framework for Python that allows you to write scripts to generate a Web App that can really work, and it supports a variety of interactive components as well.

Let me show the power of Streamlit with a real-world example.

The above Web App is a demo of what I actually developed after I know about Streamlit, with several interactive components and integration with the database, and it’s even already live. Guess how long it took me?

Less than an hour.

A beginner who can write Python can write a Web App in a very short time, and I have to say that this practical experience is truly amazing.

Moreover, the goal of Streamlit is to provide a framework for fast data visualization, so even drawing is as easy as the following.

In the Streamlit gallery there are a lot of examples that have already been done, and you can see that Streamlit covers almost all of the common components needed for daily work.

Demo Site

Let’s go back to the beginning of the example I worked on.

As we can see from the diagram, we have used six input components in the following order.

• Select box
• Calendar
• Time picker
• Text input
• Number input
• Button

The actual code is quite simple.

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

@st.cache_data(ttl=600)  
def get_user():  
    db = client.dev  
    users = db.user.find().sort('_id', -1)  
    users = [x['name'] for x in users]  
    return users  
  
user = st.selectbox('User', get_user())  
d = st.date_input('Date')  
t = st.time_input('Time')  
item = st.text_input('Item')  
amount = int(st.number_input('Amount', step=1))  
dt = datetime.combine(d, t)  
  
if st.button('Submit') and item and amount:  
    db = client.dev  
    db.accounting.insert_one({'DateTime': dt, 'Item': item, 'Amount': amount, 'User': user})  
    st.write('Submitted')

Each input component is actually a function, and the return value is the result of the input component, which makes the whole program as easy as writing a script. There is no callback and no signal, which makes the development of Web App extremely trivial.

After the script is executed, the corresponding frontend page will be generated.

Why am I so impressed?

For someone without frontend skills, making an interactive UI used to rely on packages like PyQt or tkinter.

Taking PyQt as an example, the way to write a select box is as follows.

1
2
3
4
5
6
7

def show():  
    text = box.currentText()  
    print(text)  
  
box = QtWidgets.QComboBox(form)  
box.addItems(['A','B','C','D'])  
box.currentIndexChanged.connect(show)

For components to be able to interact, they need to deal with event binding in addition to appearance definition, which is not intuitive.

Each component has its own events, and various callbacks need to be defined, all of which are learning curve. In fact, tkinter is a similar implementation.

But in Streamlit, you can do the same thing with just one line.

1
2

text = st.selectbox('Label', ['A','B','C','D'])  
print(text)

This makes development very natural. I call the component I need, and the return result is what I want.

Conclusion

The greatest advantage of Streamlit is that a person without any frontend skills (HTML, CSS and Javascript) can build a full-featured Web App in a very short time. In addition, Streamlit is a differentiator from many UI packages in providing a simple concept that makes the operation of interactive components natural, and truly what you see is what you get.

Streamlit is best suited as an internal data analytics platform or a PoC for quick validation, while its biggest shortcoming is that such a Web App is not well suited for high-traffic systems.

The main reason is because Streamlit, in order to build up the simple concept mentioned above, its implementation is to reload the whole script every time any component changes, which has a very high system load.

On the other hand, this means that the Web App is stateful and cannot be scaled horizontally through a load balancer.

And Streamlit itself so far does not support Gunicorn, so its vertical scalability is limited.

In summary, Streamlit is best suited for Data Apps, as mentioned in the official website.

A faster way to build and share data apps.

Nevertheless, for me, even though I’m able to develop frontend pages, I don’t feel I’ll ever touch HTML again. After all, who wants to deal with the extra technical stack when you can do it all in Python.

Stackademic 🎓

Thank you for reading until the end. Before you go:

• Please consider clapping and following the writer! 👏
• Follow us X | LinkedIn | YouTube | Discord
• Visit our other platforms: In Plain English | CoFeed | Venture | Cubed
• More content at Stackademic.com

Originally published on Medium

A Proven Theory of the Architect’s Difficulties

As an experienced architect, I have interviewed many candidates. But I have always had one question in my mind: why is it so hard to find candidates who meet the expectations?

Recently, after reading a study, I was able to answer this question in a more systematic way. The topic of this study is: Talent vs Luck: the role of randomness in success and failure.

It is also the 2022 Ig Informal Lecture for the Economics Prize.

Before we explain the connection between the architect’s difficulty in finding and this study, let’s quickly summarise the study.

To make a long story short, a person’s success has nothing to do with talent, but only with luck. To put it further, “Success is 1% genius and 99% luck”.

However, Thomas Edison told us that genius is 1% inspiration and 99% perspiration. In other words, to be successful, effort is only 0.99%.

Why are so few candidates qualified to be architects?

Because being an architect is not something you can do just by working hard, it’s very much a matter of chance.

Wait, that’s a bit arbitrary, isn’t it?

In fact, every single one of those interviewed was a good senior engineer — yes, every single one.

They all excel in their positions and have a good grasp of what they are responsible for. Even a team leader or manager who hadn’t actually written code in a long time still had a deep understanding of the service they were working on.

However, these candidates were eliminated.

The reason was simple: not enough experience. Yes, I agree that they can all fully handle one or a few microservices. But the volume of these microservices was just not there, either because they lacked the volume of traffic and high-frequency reads and writes, or because they lacked the complexity and interaction with other systems.

This kind of experience is no match for an architect of a high-volume e-commerce platform.

Did you notice a key point? They all do their jobs well or even brilliantly, but they are still inexperienced.

Specialist vs. Generalist

This is actually the difference between I-shaped talents and T-shaped talents, i.e. Specialist and Generalist.

If you want to become a qualified architect, it is not enough just to do a good job. But to be honest, even if you work very hard, you may not be qualified. As I mentioned at the beginning, whether you can become an architect or not is mostly a matter of luck.

• If you are lucky enough to have worked for a large enterprise, then there is a little more possibility.
• If, luckily, the company is willing to invest in various training, then there are more possibilities.
• If you are lucky enough to have friendships with architects, then you have a little more possibilities.

Can’t we learn the skills we should have by our own efforts?

Yes, but it’s really hard.

If you don’t see how wide the world is, you can’t imagine what you need to learn.

There is a Chinese idiom, “Frog in the bottom of a well”. A frog who lives in a well all his life cannot know the outside world.

Back to the architect’s perspective, let’s look at a few real-world examples.

How do you know you need to learn caching if a system works well with one database and one API service?

How do you know you need to learn sharding if the amount of data can always be carried by a single database?

How do you know you need to learn message queuing if the logic of the system is simple and can be responded in a short time?

More specifically, if a system does not have any search requirement, how do you know you need to learn Elasticsearch?

All these are the vision that an architect needs to have, so do you know why most of the time you need to rely on “luck”?

Even so

There are still possibilities for us to try our best.

• Read more books.
• Attend more seminars.
• Visit more blogs.

Even if we don’t know what’s missing, we can still become what we want to be by learning through various channels.

Architects are on a never-ending journey of learning.

Conclusion

We’ve talked a lot about what it takes to be an architect, and explained why it’s so hard to find a qualified candidate, but I have to say that being an architect is just one of the “choices” in a career.

It doesn’t mean an architect is superior, it doesn’t mean an architect is invincible, and it doesn’t mean an architect is better than a senior engineer.

If learning is your hobby, then architect is a good choice.

But a senior software developer is also a good choice.

Originally published on Medium

Netflix的首頁是經過客製化的，會根據使用者的喜好有不同的外觀，當然，推薦影片的種類也會完全根據使用者的偏好。要做到這種程度，是因為在系統背後有一個應用程式會根據各種演算法將好的UX遞送給使用者。

這個應用程式稱為RMI (Real-time Merched Impression)，其架構如下。

graph TD
    imp[Impression Source] --> kb1[KeyBy] --> join[Join] --> rmi[RMI Sink]
    pb[Playback Source] --> kb2[KeyBy] --> join

Generating Avro Schemas from Go types

To implement auto data ingest, reflect go struct into Avro

Previously, I have introduced the evolutionary history of data infrastructure, and I have also introduced real-time data analysis in detail.

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

Both of these articles mention a key player, Debezium. In fact, Debezium has had a place in the modern infrastructure. Let’s use a diagram to understand why.

This architecture diagram should be consistent with the data ingest path of most data infrastructures. In order to store and analyze data in a unified way, centralizing the data in a data warehouse is a general solution.

Debezium captures data changes from various source databases and then writes them to the data warehouse via Kafka and Kafka’s consumers. In other words, Debezium is also a Kafka producer.

This architecture looks fine, once there are changes in the data will be captured by Debezium and eventually written to the data warehouse, but there is actually a difficult problem behind needs to be solved, that is, the database schema. In order to provide data analytics, the data warehouse is “schemaful”, but how to make the schema of the data warehouse align with the schema of each source database?

The good news is Debezium supports schema registry, it can capture DDL changes from source databases and synchronize them to the schema registry so that Kafka consumers can get schema from the schema registry and synchronize the schema in the data warehouse.

Let’s make the architecture mentioned above even more complete.

The most common format for describing schema in this scenario is Apache Avro.

Unfortunately, a schemaless database like MongoDB doesn’t have DDL at all, so of course Debezium can’t capture DDL, then how does Debezium upload the schema in this scenario? The answer is to guess, by getting the data to determine the data type, and write the type into the schema registry.

As you can imagine, this is very risky, and the data warehouse schema often faces compatibility challenges due to such guessing. Therefore, there needs to be a way to proactively upload the schema so that the Kafka consumer can correctly recognize the data type and correct the schema of the data warehouse.

Knowing it intellectually is one thing, but knowing it emotionally is quite another.

The story is this: our microservice is developed in Golang and uses mgm as the ORM for MongoDB.

In Golang, we use struct as a Data Transfer Object (DTO), and in the mgm scenario, these DTOs can be marshaled as JSON. So, what we need to do is to generate the corresponding Avro schema based on the definition of struct.

Therefore, I made the following go module.

The use of this go module is quite simple.

If we have a struct as follows.

1
2
3
4
5
6
7

type Entity struct {  
	AStrField    string  `json:"a_str_field"`  
	AIntField    int     `json:"a_int_field"`  
	ABoolField   bool    `json:"a_bool_field"`  
	AFloatField  float32 `json:"a_float_field"`  
	ADoubleField float64 `json:"a_double_field"`  
}

Then just import the module to generate the corresponding Avro.

1
2
3

import "github.com/wirelessr/avroschema"  
  
avroschema.Reflect(&Entity{})

The result will be an Avro JSON.

1
2
3
4
5
6
7
8
9
10
11

{  
    "name": "Entity",  
    "type": "record",  
    "fields": [  
      {"name": "a_str_field", "type": "string"},  
      {"name": "a_int_field", "type": "int"},  
      {"name": "a_bool_field", "type": "boolean"},  
      {"name": "a_float_field", "type": "float"},  
      {"name": "a_double_field", "type": "double"}  
    ]  
}

In addition, the module also supports various complex types and nested types as follows.

• array and slice
• map
• struct in struct
• time.Time
• array of struct
• etc.

Of course, the omitempty tag used by JSON in Golang is also supported.

Fields marked with omitempty become Avro’s union type, which is an optional field. Let’s take a more complicated example.

1
2
3

type Entity struct {  
	UnionArrayField []int `json:"union_array_field,omitempty"`  
}

If it is an optional int array, then it will become a union type with null when converted to Avro.

1
2
3
4
5
6
7
8
9

{  
    "name": "Entity",  
    "type": "record",  
    "fields": [  
      {"name": "union_array_field", "type": ["null", {  
        "type": "array", "items": "int"  
      }]}  
    ]  
}

More detailed tests can be found in the unit test file: reflect_test.go.

However, this is not enough for us, because we need to support various special types in mgm, such as primitive.DateTime or primitive.ObjectID, and most especially mgm.DefaultModel.

Therefore, it is necessary for this module to provide customization capabilities to extend support for more special types, thus, a plugin is provided for this module to be used. To support mgm, you can use the already made submodule.

The following is an example of a struct that uses the mgm type.

1
2
3
4
5
6
7
8
9

type Book struct {  
	mgm.DefaultModel `bson:",inline"`  
	Name             string             `json:"name" bson:"name"`  
	Pages            int                `json:"pages" bson:"pages"`  
	ObjId            primitive.ObjectID `json:"obj_id" bson:"obj_id"`  
	ArrivedAt        primitive.DateTime `json:"arrived_at" bson:"arrived_at"`  
	RefData          bson.M             `json:"ref_data" bson:"ref_data"`  
	Author           []string           `json:"author" bson:"author"`  
}

To support these special types, you can use the built-in custom MgmExtension. This works just like the original Reflect, except it requires the use of a customized instance.

1
2
3
4
5
6
7
8
9

import (  
	"github.com/wirelessr/avroschema"  
	"github.com/wirelessr/avroschema/mongo"  
)  
  
reflector := new(avroschema.Reflector)  
reflector.Mapper = MgmExtension  
  
reflector.Reflect(&Book{})

There are a couple of important points to note here.

1. DefaultModel is a type that directly generates three fields, _id, created_at and updated_at.
2. If primitive.ObjectID then it is treated as a string, since the UUID mentioned in Avro specification is also treated as a string.
3. If primitive.DateTime, then it is treated as time.Time.
4. A special case is bson.M, which is treated as a JSON string because there is no way to be sure what is in it.

According to the above rules, the final Avro schema is as follows.

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

{  
	"name": "Book",  
	"type": "record",  
	"fields": [  
		{ "name": "_id", "type": "string" },  
		{ "name": "created_at", "type": "long", "logicalType": "timestamp-millis" },  
		{ "name": "updated_at", "type": "long", "logicalType": "timestamp-millis" },  
		{ "name": "name", "type": "string" },  
		{ "name": "pages", "type": "int" },  
		{ "name": "obj_id", "type": "string" },  
		{ "name": "arrived_at", "type": "long", "logicalType": "timestamp-millis" },  
		{ "name": "ref_data", "type": "string" },  
		{ "name": "author", "type": "array", "items": "string" }   
	]  
}

Conclusion

When implementing the requirement to convert Golang struct to Avro schema I first tried to look for existing packages, but unfortunately I couldn’t find any. However, there is a similar project that aims to convert Golang struct to JSON schema, so I referred to his practice and made a version of Avro schema.

In fact, there are still some specifications in the Avro schema that I don’t support, such as the enum type, and there are also many optional parameters that I don’t support because they don’t correspond to struct, such as namespace, aliases, etc.

But for now, this implementation satisfies my need for data ingest, and allows Kafka consumers to have the correct schema to work with. Maybe I’ll continue to improve this project sometime in the future so that it can really support all kinds of Avro specifications. But I have to say, in terms of functionality, it’s pretty useful right now.

The project has full unit testing and linter, and I’m sure the code structure isn’t too complicated, so feel free to contribute.

Stackademic

Thank you for reading until the end. Before you go:

• Please consider clapping and following the writer! 👏
• Follow us X | LinkedIn | YouTube | Discord
• Visit our other platforms: In Plain English | CoFeed | Venture

Originally published on Medium

Optimizing Elasticsearch Reindex without Downtime

A Guide to achieving zero downtime, high efficiency, and successful migration updates

When using Elasticsearch, there are always times we need to modify the index mapping, and when it happens, we can only do _reindex. In fact, it is a pretty expensive operation because it can take up to several hours to make a complete replication of an index, depending on the amount of data and the amount of shards.

The time spent is not a big problem, but more seriously, it can affect the performance and even the functionality of the production environment.

I believe we all understand that data migration consumes a lot of hard drive resources, it definitely affects performance, but what about functionality?

Let’s take a regular _reindex as an example. Suppose we have created an alias on the index. If we don’t have an alias, we’re in big trouble.

A regular _reindex procedure is divided into two steps.

1. Call the _reindex command to start data migration.
2. When the data migration is complete, call the _aliases command to switch between the old and new indexes.

After step 2, the new index is officially operational, and will be responsible for all read and write requests. However, this is a perfect ideal scenario, in reality, things won’t work out that way.

Here is a normal scenario.

Actually, during the data migration period or before switching aliases, the client will continue to write data to the original index, and these new changes will not be migrated to the new index, which leads to data inconsistency.

For the client, the feeling is that after changing the alias, all the changes just made will disappear. Furthermore, as I just mentioned, a big index migration can take hours, so the client’s feeling must be obvious.

So what to do?

The correct flow of Reindex

The above flow makes two changes to the original flow.

1. _reindex must use external type.
2. _reindex is needed again after switching aliases.

Let’s explain the concept of the external type.

By default, _reindex is internal, and such data migration is done by using the original index to overwrite the new index anyway, and dropping the _version of the documents, so all documents in the new index start over.

If using the external type, the _version of the documents will be carried over to the new index during data migration, then the _version will be compared if there is a conflict between the _id of the old and new indexes. Only if the version of the original document is greater than the destination document will be overwritten.

A bit abstract? Let’s take an example.

Let’s say the original index has a document like the following, with Elasticsearch metadata at the beginning of the underscore.

1
2
3
4
5

{  
    "_id": "1",  
    "_version": 1,  
    "data": "Hello Elastic"  
}

When we do external data migration, _version: 1 is also written to the new index. if someone changes the original document to Hello Search during the data migration period, then the complete document will look like the following.

1
2
3
4
5

{  
    "_id": "1",  
    "_version": 2,  
    "data": "Hello Search"  
}

A redo of _reindex will find 2 > 1, so it will be overwritten with Hello Search.

Then, what if the second _reindex has someone modifying the documents in the new index? For example, if someone changes Hello Elatic to Hello Elasticsearch in the new index, will it be overwritten with the old value? The whole process would look like below.

The answer is no, because the original version has to be greater than the new version to be overwritten, if they are equal (both 2) then they are not affected.

One more problem.

Although we will do a second _reindex to patch the data, if the patch time is very long, it will still be inconsistent for the users. There are two ways to shorten the reindexing time.

1. Minimize the time of the first _reindex as much as possible.
2. Filter the patch data in advance.

Regarding the first point, the _reindex process is controlled by Elasticsearch, what else can we do to improve efficiency? Hey, there is.

We can modify the settings of the new index to minimize the IO overhead during data migration.

1. refresh_interval = -1
2. number_of_replicas = 0

It’s quite straightforward. First of all, the purpose of turning off refresh_interval is to allow the data migration period to just focus on writing to the Trans log and not spend extra disk IO on Lucene.

Secondly, turning off number_of_replicas reduces the additional data replication overhead that the cluster has to deal with.

On the other hand, in addition to reducing the time of the first _reindex, some data filtering can be used to reduce the amount of data for the second _reindex.

For example, bringing in the last update time of the data during the _reindex is a possible solution. Assuming that each document has an updated_at field, then adding the following condition to the query of the _reindex would be effective.

1
2
3
4
5

{  
    "range": {  
        "updated_at": { "gte": "now-1d"}  
    }  
}

Conclusion

Based on the above mentioned details, let’s list out the ideal flow of a reindex.

1. Create the destination index.
2. Update the settings of the destination index. (refresh_interval = -1 and number_of_replicas = 0)
3. Use external type for _reindex.
4. Switch aliases from original index to destination index.
5. Do _reindex again using external type, preferably with additional filtering.
6. Update the destination index setting again. (refresh_interval = null and number_of_replicas = null)

According to the official document, setting to null allows to restore the original setting.

Because _reindex is unavoidable, it is important to know how to do _reindex without downtime.

In fact, with Elasticsearch’s streaming index, there are more elegant ways to get it done. However, there are a lot of limitations on the use case for a streaming index, so it’s more common to use a regular index in practice.

This article provides a complete procedure to do _reindex as fast as possible and minimize the time of data inconsistency. However, all of this assumes that the alias is properly created, and if it is not, then more extra steps are required. I feel the lack of aliases is already a violation of Elasticsearch’s best practices, so this article doesn’t specifically cover that scenario.

Stackademic

Thank you for reading until the end. Before you go:

• Please consider clapping and following the writer! 👏
• Follow us on Twitter(X), LinkedIn, and YouTube.
• Visit Stackademic.com to find out more about how we are democratizing free programming education around the world.

Originally published on Medium

Getting Started with Flink SQL, Apache Iceberg and DynamoDB Catalog

Navigating Flink SQL, Apache Iceberg, and DynamoDB Catalog for a Lakehouse Experience

Due to the technology transformation we want to do recently, we started to investigate Apache Iceberg. In addition, the data processing engine we use in house is Apache Flink, so it’s only fair to look for an experiment environment that integrates Flink and Iceberg.

The purpose of this experiment is simply to experience using Flink SQL to operate Iceberg, but I didn’t realize that it would be so difficult to set up the whole environment.

Actually, Iceberg’s official document has a “large” description of the integration with Flink, but it is based on Hadoop and Hive. For a data engineer who has not experienced the legacy Hadoop ecosystem, Hadoop is like an abyss, and obviously the official documents don’t work.

So I referred to Dremio’s article, fixed a few fatal flaws in the article, and finally built an experiment environment where I could experience the full Flink Iceberg integration. Moreover, it’s completely local and free of charge.

Before introducing the environment, let’s quickly introduce Iceberg.

A short introduction to Iceberg

Apache Iceberg is one of the three types of lakehouse, the other two are Apache Hudi and Delta Lake.

Lakehouse, in a nutshell, is to provide a layer of interface for Object Storage that can be operated by SQL, so that lakehouse has more structured data than data lake and has the ease of use of a data warehouse.

In addition, lakehouse also provides integration with major common data processing engines, so we can enjoy the low cost of Object Storage but at the same time make the data easy to use.

Here’s a diagram that I found very impressive.

Comparison of the three types of lakehouses is not the point of this article, there is a detailed benchmark in Kyle Weller’s article.

Experiment environment

After reading the above introduction, we understand in order to be able to perform structured data operations on Object Storage, we need a mechanism to store a schema, which is called a catalog on Iceberg.

Iceberg provides a number of options for implementing catalogs, including the Hive metastore mentioned at the beginning, and DynamoDB, which will be used in this article.

The entire playground is completely free, and at its core are the following three components.

1. DynamoDB: This is the catalog store, and uses the local version to avoid expenses.
2. Minio: This is where the actual iceberg is stored, again using minio to avoid the expense of s3.
3. Flink: the key to Playground.

The entire experiment environment has been in Github, you can clone to follow the steps.

The journey of experimentation

This environment is extremely easy to use, all you have to do is docker-compose up -d and you’re good to go.

But to get it working, I looked at a lot of articles and found that none of them actually worked, either they were buggy or just provided snippets, so I set out to make one myself.

I especially want to talk about Dremio’s article, it inspired me a lot. The only difference is that it uses Nessie as the catalog, and I’m not familiar with Nessie at all, so I used DynamoDB instead.

On the other hand, about DynamoDbCatalog, there is poor information on the Internet, not even what parameters need to be modified, I finally found out by reviewing the following source code.

Changing DynamoDB’s table name should be important if the environment goes to production, but even this parameter (dynamodb.table-name) has to be found in the code.

In addition, because the experiment environment enabled minio as an alternative to s3, there is another parameter besides endpoint that is also critical, and it requires a change in the path style approach, and again, this parameter (s3.path-style-access) is what I found by looking at the source code, which is completely undocumented.

Finally, the experiment environment has been built, there is a docker image that has been packaged with Flink and all the dependencies that need to be built, the image is a bit large and needs to be built for a period of time, so I also provide a ready-made one.

Just replace the image used by sql-client in docker-compose.yml.

Originally published on Medium

Agile Evolution: Beyond Methods

Why the Agile Manifesto author said Agile is dead

Recently, I reviewed “Agile is dead” by Dave Thomas, one of the authors of the Agile Manifesto, and after a few years of reviewing it again, I still have some new insights because of my different role (from developer to architect).

While the topic is “Agile is dead,” a more accurate statement is that Agile has survived in a wrong form. Agile hasn’t died; it’s merely subsisting.

Back to the root of the problem, Agile is essentially a spirit, a concept, or more precisely, an adjective rather than a noun. Thus, the essence of an Agile development process lies in the development process itself, not in Agile methodologies. The key is not in frameworks like Scrum, Kanban, or various other methodologies but in how to get things done quickly.

With this understanding, we realize that achieving speed in work varies based on organizational structure and existing practices. Therefore, any “general rules” provided by Agile experts or books lack reference value because:

• No rules are universal.
• All rules need context.

There are no one-size-fits-all rules; Agile practices should be adjusted according to the needs of the organization. Agile is not something that can be achieved by hiring a consultant for a short period of time, the key is to find a reasonable direction of evolution by grasping the spirit of Agile.

The conclusion of the entire talk can be summarized as follows:

• Agile is not about what you do.
• Agility is about how you do it.

Translated into concrete steps (not specific processes), it might be the following process:

1. Identify the problem.
2. Review historical experiences.
3. Make small adjustments and observe the results.

In fact, this aligns the familiar DevOps’ The Three Ways.

Additionally, the talk introduces another compelling point, especially given my current role as an architect:

• Don’t let the turkeys get you down.
• All experts are turkeys.

When I was a developer before, I used to dislike authoritative figures coming in to give directions and then leaving after stirring up the situation. They were always clean-cut, directing those in the mud from the safety of the shore. Thus, now as an architect, I try to avoid making the same mistake.

Most of the time, I provide a few suggestions and highlight the pros and cons of various solutions based on the context, allowing the engineering team I collaborate with to have a blueprint. But they need to weigh it against their own measurements themselves. Because, in the end, I am also a turkey.

This also echoes the commonly mentioned chicken and pig story in Agile.

Originally published on Medium

這篇是翻譯Netflix最近釋出的Psyberg系列文，同時也會補充一些相關的資訊。

最主要的出處是以下三篇：

• Streamlining Membership Data Engineering at Netflix with Psyberg
• Diving Deeper into Psyberg: Stateless vs Stateful Data Processing
• Psyberg: Automated end to end catch up

Navigating Breaking Changes with Unified Queries and Gradual Schema Upgrades

When we are using a relational database, we often use incremental ID as the primary key (e.g. AUTO_INCREMENT in MySQL or SERIAL in Postgres), have you ever thought about what will happen if the incremental ID is used up?

For example, Postgres raises the following error.
ERROR: nextval: reached maximum value of sequence "<table_name>_<pk>_seq".

Don’t feel hard of happening, it is possible to accidentally set the primary key to SERIAL instead of BIGSERIAL, and then a value larger than 2147483647 will overflow. For a database, it’s not a big number.

If such a problem unfortunately occurs, what should we do to solve this?

There are a lot of articles on how to migrate data from the original table to a new table, but in fact, I don’t recommend it. Of course, it’s easiest to migrate all the data to a new table at once, but there are several problems like the following.

1. Migrating a large volume data consumes a lot of database resources, and will have a significant impact on the production environment.
2. Data inconsistency is inevitable during the migration process. For example, if data is migrated to a new table and then a new version of application is deployed to read and write the new table, if there is data written to the old table during the migration period, then there will be inconsistency.
3. If there is a downstream application that listens for updates to the source database (or CDC), then the downstream application will be flooded with events.

Therefore, I recommend that applications support both old and new tables. Creating new data is always written to the new table, but reading or updating is done according to whether the data is in the old table or the new table.

But how? For querying, should I look for the new table first and then the old table as in the example below?

1
2
3
4
5
6

if (ret := find_orig_table(params)) is not None:  
    return ret  
elif (ret := find_new_table(params)) is not None:  
    return ret  
else:  
    return None

Actually, there is an even simpler approach.

Solution Detail

To better describe the solution, let’s prepare the test environment with Postgres.

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

CREATE TABLE table_with_auto_increment (  
    id SERIAL PRIMARY KEY,  
    common_column1 VARCHAR(50),  
    common_column2 INTEGER,  
    common_column3 TEXT  
);  
CREATE EXTENSION IF NOT EXISTS "uuid-ossp";  
CREATE TABLE table_with_uuid_primary_key (  
    id UUID DEFAULT uuid_generate_v4() PRIMARY KEY,  
    common_column1 VARCHAR(50),  
    common_column2 INTEGER,  
    common_column3 TEXT  
);

Above is an old table (table_with_auto_increment) with SERIAL as primary key, it is easy to overflow, so we want to replace it with a new table (table_with_uuid_primary_key) with UUID as primary key.

There is a lot of debate on whether UUID as primary key is good or bad, but that’s not the point of this article, it’s just one of the examples.

Based on such a table structure, it will have a DTO (data transfer object) similar to the following.

1
2
3
4
5
6
7

from dataclasses import dataclass  
@dataclass  
class DTO:  
    id: ...  
    common_column1: ...  
    common_column2: ...  
    common_column3: ...

The DTO will be the same whether using the new table or the old table, so the business logic does not have to be modified during the migration process.

Then, what’s the problem of looking for the new table first and then the old one? The main problem is twice accessing the database is two round trips, which leads to one more expense, and secondly, the two functions also have maintenance complexity. Therefore, it would be great if there is a way to use one database access to assemble a DTO from two tables.

The first idea was to use UNION. However, it is worth mentioning that UNION requires both tables to have exactly the same datatype, and the id of our old and new tables are of different types.

Therefore, we need to convert the id type at the same time with UNION, and converting both of them to text is the simplest way. This is done in the following example.

1
2
3
4
5
6
7

SELECT id::text, common_column1, common_column2, common_column3  
FROM table_with_auto_increment  
WHERE common_column1 = 'test' AND common_column2 = 123  
UNION ALL  
SELECT id::text, common_column1, common_column2, common_column3  
FROM table_with_uuid_primary_key  
WHERE common_column1 = 'test' AND common_column2 = 123;

Our search criteria is common_column1 is test and common_column2 is 123, so we search both tables directly and combine the results. In general, only one of the two tables will have data ( with no bugs ), so it is very efficient to create the DTO in this way.

In this example, the purpose of creating two tables is to change ids seamlessly, so we can simplify the process through type conversion.

However, there are other scenarios where our functionality iteration leads to breaking change, so we make two tables to allow the old application to gradually transition to the new application, it may not be possible to simply do type conversion (maybe the types are not compatible at all).

So, we can consider to use OUTER JOIN method, and use the common field as the key of JOIN, so that the result will be the union of the fields of the two tables.

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

SELECT  
    t1.id AS t1_id, t2.id AS t2_id,  
    COALESCE(t1.common_column1, t2.common_column1) AS common_column1,  
    COALESCE(t1.common_column2, t2.common_column2) AS common_column2,  
    COALESCE(t1.common_column3, t2.common_column3) AS common_column3  
FROM  
    table_with_auto_increment t1  
FULL OUTER JOIN  
    table_with_uuid_primary_key t2  
ON t1.common_column1 = t2.common_column1 AND t1.common_column2 = t2.common_column2  
WHERE  
    t1.common_column1 = 'test' AND t1.common_column2 = 123  
OR  
    t2.common_column1 = 'test' AND t2.common_column2 = 123;

The above example is very similar to UNION, we need to find out common_column1 is test and common_column2 is 123 in two tables, but we need to list the remaining columns.

Nevertheless, the whole SQL is a bit annoying, such as the series of COALESCE, so I suggest some template rendering to make the development more intuitive.

Here’s an example of using Python with jinja2.

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

from jinja2 import Template  
columns = ['common_column1', 'common_column2', 'common_column3']  
template_str = """SELECT  
    t1.id AS t1_id, t2.id AS t2_id  
{% for column in columns %}  
    , COALESCE(t1.{{ column }}, t2.{{ column }}) AS {{ column }}  
{% endfor %}  
FROM  
    table_with_auto_increment t1  
FULL OUTER JOIN  
    table_with_uuid_primary_key t2  
ON t1.{{condition_column1}} = t2.{{condition_column1}} and t1.{{condition_column2}} = t2.{{condition_column2}}  
WHERE  
    t1.{{condition_column1}} = '{{condition_value1}}' AND t1.{{condition_column2}} = {{condition_value2}}  
OR  
    t2.{{condition_column1}} = '{{condition_value1}}' AND t2.{{condition_column2}} = {{condition_value2}};  
"""  
condition_column1 = 'common_column1'  
condition_column2 = 'common_column2'  
condition_value1 = 'test'  
condition_value2 = 123  
template = Template(template_str)  
query = template.render(  
    columns=columns,  
    condition_column1=condition_column1, condition_column2=condition_column2,  
    condition_value1=condition_value1, condition_value2=condition_value2  
)  
print(query)

At this point, we’ve been able to make the transition from the old table structure to the new one seamless for the application.

Conclusion

Although in this article we have introduced how to solve the problem of id exhaustion, it can be used in any scenario where we need to make a breaking change to the table structure of the database.

When the data volume is small, breaking change is not a big deal, we just need to migrate all the old tables to the new ones, but when the data volume is huge, such an approach will increase a lot of operation risks.

A complete table migration involves not only the data migration itself, but also how to keep the production environment unaffected. The most common problem we encountered is how to synchronize the data to the new table if there are still clients writing to the old table during the data migration period.

In addition, in order to avoid affecting the production environment, we usually have to choose the off-peak time to perform the migration, which inevitably requires many departments to work overtime together.

For instance, the development team needs to release a new version of the program, the operation team needs to migrate the data, the SRE team needs to monitor the risk of online, and the QA team needs to validate. It is not worth for me to work so hard just for the sake of table migration.

Therefore, it’s best to be able to upgrade your application to a new version seamlessly. This article introduces a simple but effective practice to avoid many unnecessary breaking changes. Of course, there are many different ways to achieve this, if you have some good solutions, please feel free to share them with me.

Stackademic

Thank you for reading until the end. Before you go:

• Please consider clapping and following the writer! 👏
• Follow us on Twitter(X), LinkedIn, and YouTube.
• Visit Stackademic.com to find out more about how we are democratizing free programming education around the world.

Originally published on Medium