We can only have two of the three, and the price of being fast and good is expensive. Is there any way to do it as fast, as good and as cheap as possible? I believe the following article will be helpful.
Apptio is an enterprise IT management solutions company with several programs for managing agile development processes. Acquired by IBM in June 2023, Apptio went from delivering a promising product to incubation and graduation in just a few years, itself a very agile company.
They wrote the above methodology on how to accelerate software development, and it’s a great read as it paints a detailed picture of the various software development scenarios.
Here’s a quick summary of some of the key points.
The blue blocks are our ultimate goal, to speed up development.
The red blocks are those that slow down development, and of course, the fewer the better.
The green blocks indicate items that will increase development speed, i.e. the more the better.
The yellow blocks mean that the right amount of these items will increase the speed, but too much will slow it down.
So, let’s look at the picture and tell the story.
We should:
Reduce system complexity
Reduce re-work
Focus on work
Reduce non-value added activities
Do refactoring properly
And so on.
But what exactly should be practiced? The article explains the details, so I won’t write it all out, I recommend to read the article carefully, I feel it is well worth reading!
Navigating the Path to Clean Code in Real-world Scenarios
How to design and implement clean architecture more easily has been described in detail in my previous articles. If you are interested in the details, you can refer to the following series of articles.
Instead of deeply diving into how to design and implement a clean architecture, this article will answer a question that we often encounter.
When should I consider clean architecture?
The main reason for asking this question is we often don’t design code with a clean architecture mindset from the start, but rather decide at some point to do a refactoring and introduce clean architecture.
This is a natural process because we don’t always have the ability to figure out how to encapsulate the domain in the first place, and we don’t always have the time to have a complete design at the beginning.
So, when should we refactor? And how? These two questions will be the core of this article.
Case Study
To answer these two questions we need a practical example, so let me describe my thought process using a chatbot.
Suppose I want to use a chatbot to implement an accounting program. The reason why I chose chatbot is to avoid complicated interface interactions, I just need to talk to my usual IM App to achieve the goal of accounting.
The overview of chatbot is as follows.
The user sends commands to the chatbot through a conversation, and the chatbot receives the message, encapsulates it and sends it to a web service via a webhook; the web service extracts the content with the corresponding SDK, processes the commands, and then sends the results back to the user. The process of processing the message involves accessing the database.
The code flow of the app is organized in the following five steps.
Extract: Use SDK to extract the received request.
Parse: Parses the message into a corresponding command.
Handle: Execute the command and generate the result. Interaction with the database also occurs at this stage.
Encapsulate: After handling the request, the result is packaged into the correct format by the SDK.
Send: Finally, the result is sent back to the user.
The whole core process will be the second and third steps. Without considering the clean architecture, the entire code would look like the following.
1 2 3 4 5 6 7
if msg == 'ooo': ret = user_guide() elif msg.count('x') > 10: ret = insert_items() elif'oxo'in msg: ret = get_report() # and so on
By parsing the msg, we know the command and take the corresponding action, in other words, this is the business logic of the whole program.
In the above example, we have three kinds of commands and three handlers, which correspond to the three common functions of a accounting program.
ask how to use the program.
write in accounts.
generate reports.
I believe that adding a fourth command would be too painful for some people, and I would start thinking about refactoring it at about this point.
So to answer the first question: when should I consider refactoring with clean architecture?
The answer is simple: when the requirements keep iterating and the code becomes difficult to maintain.
But how to refactor? How to introduce clean architecture?
Refactoring
According to the above description, we have two main business logics.
Parser
Handler
Let’s review the onion architecture one more time.
Calling chain has to go from the outside to the inside, and the most inside is business logic.
So I will plan the refactored components as shown below.
Let me explain these components.
Controller: Handles the in/out of the webhook and gets the actual message and user id of the caller with the SDK.
Service:
Call Factory to get concrete command.
Execute Command directly and get the result, may be a string or an exception.
Factory: Handles the logic of generating the Command, i.e. the command parser.
Command: Handles the actual business logic, i.e. the handler, and is responsible for converting the result of the repo to a string.
Repo: handles database calls and wrapping the database format into a DTO, no logic involved.
Splitting the domain in this way has several advantages.
Factory unit tests only need to have msg and match Command type, the implementation is quite simple.
Command encapsulates the full behavior of individual commands and isolates the database implementation, so unit testing can easily replace the database through dependency injection and just verify business logic.
Repo provides an interface to the database and wraps the format of the manipulated data into a DTO, providing a common specification.
Back to the onion architecture, Factory and Command are considered business logic, i.e., Entity, and Service is responsible for calling the Entity in order and doing the corresponding error handling, which belongs to the middle Use Case layer. The outer Controller handles the SDK and web framework, and calls the Service to get the result.
The only thing need to discuss is the outermost layer of DB, in the class diagram is actually Mongo Impl, through the implementation interface to make it out of the bottom layer of the calling chain, all the classes will only touch the Repo rather than the implementation of MongoDB. How to explain the source of the calling chain, i.e. the outer layer, is the DB?
The reason is that the outer layer of the code needs to prepare a concrete repo of the Mongo Impl and pass it to the Controller. In my current implementation, I’ve changed my approach a little bit by passing the class type instead of instance.
A similar implementation can be found in my previous Golang tutorial.
Then the “Command”, where the actual business logic is executed, and the following examples will only pick one of them as a demonstration.
1 2 3 4 5 6 7 8 9 10 11 12
classInsertManyCommand(Command): defexecute(self, repo_cls): tokens = [x.strip() for x in self.msg.split(',')] data = [] for token in tokens: item, cost = [t(s) for t,s inzip((str,int), token.split())] data.append(Item(item, cost)) repo = repo_cls(self.uid) success_cnt = repo.insert_many(data) returnf'Successfully wrote {success_cnt} record(s)'
Conclusion
Let’s go back to the title, when should we start considering clean architecture? I feel the answer is obvious, when we need to change the code but we don’t know where to start, that’s a good time to do it.
Why do I let my business logic consist of Factory and Command? Because once I’ve clarified the behavioral patterns of the bot, I can clearly recognize that two things are necessary: parsing and handling, so these two corresponding entities are created. In addition, when these two things are separated, the difficulty of unit testing is significantly reduced.
I can fully test parsing messages, and of course I can test every self-contained command completely.
When we mention clean architecture, will always make people think of hexagonal architecture or domain-driven development and other formal rules, but in practice to make the architecture clean, as long as the basic essentials will be enough, that is, the concept of the onion architecture.
Some of the design patterns used in this article, such as the factory method or strategy pattern. Even though I didn’t follow the textbook, I was able to achieve a good result.
When I drew that class diagram, I didn’t actually have a specific design pattern in mind, but rather, it was based on what shape I wanted the code to be organized into. I first thought about how I would like to decouple and individually test a command if I needed to add one, and then I created two components (Factory and Command) with this idea in mind.
Although Factory has a relatively simple role, I indeed thought for a while about what context Command would be responsible for. Take InsertManyCommand for example, there are these options.
The Service is responsible for tokenization and conversion to DTO.
The Repo is responsible for tokenization and writing directly to the database.
The Service handles the presentation of the result string.
All of these were eventually put into Command, because how the strings are split and how the results are presented should be part of the business logic, and naturally should be handled by Command, which is the core of the business logic.
Once we have a concrete idea, we can fill in the details of the class diagram, and then we just need to follow the diagram to write the program exactly as it is designed.
Lastly, my thoughts on ORMs. In fact, I don’t use ORMs to implement database access, but rather use pymongo with a custom DTO.
Why don’t I just use an ORM? The reason is that ORM will make developers overlook the existence of database and make clean architecture difficult to achieve.
From the onion architecture, we know the database is on the outer layer and should not be called by any object, but with ORM, we can easily develop the ORM as an entity into the inner layer of the onion architecture, and it even contains a lot of business logic. This is difficult to test and difficult to maintain, and is the worst situation of all.
Explaining Python 3.10’s new features with real-world examples
Pattern matching has finally been supported in Python since 3.10, and this feature, which is common to many functional programming languages, has been painlessly ported to Python.
However, for those of us who are new to pattern matching, it’s hard for us to figure out the right use cases and thus the code we write is not much different from an if-else, with at most a little more syntactic sugar in the if-else to allow for more fine-grained definitions of the variables.
To understand the real use case of pattern matching, we have to review the paragraph in PEP 635.
Much of the power of pattern matching comes from the nesting of subpatterns. That the success of a pattern match depends directly on the success of subpattern is thus a cornerstone of the design.
Nesting of subpatterns sounds a bit abstract, what are some practical examples of this property?
The easiest data structure to understand is the tree.
Taking a binary tree as an example, traversing from the top to the bottom, each node actually satisfies the following properties.
Each node will have an attribute of its own, as well as a child to the left and right, and this is the typical nested structure. Therefore, one of the most practical examples is to look for tree nodes that match the pattern.
AST Tree
AST tree is a very common method used for lint or syntax checking.
Suppose we have a requirement to check whether a source code has a direct call to print, then we can use if to compare the conditions and find out if the node that matches the condition violates the lint rule.
1 2 3 4 5 6 7 8 9
for node in ast.walk(tree): if ( isinstance(node, ast.Call) # It's a call and isinstance(node.func, ast.Name) # It directly invokes a name and node.func.id == 'print' # That name is `print` ): sys.exit(0) sys.exit(1)
From the above code we know we want to match a node ast.Call and it is a call function and the function name is print, three conditions in total.
Now, we know the pattern conditions we want to compare, let’s rewrite it as follows using pattern matching.
1 2 3 4 5 6 7
for node in ast.walk(tree): match node: # If it's a call to `print` case ast.Call(func=ast.Name(id='print')): sys.exit(0) sys.exit(1)
With pattern matching, we can make the conditions more straightforward by comparing them to the class definition we have in our head.
From the code we can see pattern matching makes comparing conditions simple, but that’s not the power of pattern matching, because we haven’t compared nested structures yet. Let’s look at the next example.
Parsing Tree
Suppose now I want to find some Patterns on a binary tree with various colors and shapes.
Given a tree of this kind, I want to know:
Is there a green square on the tree with a red circle attached to the left node?
Does the tree have any shape of any color, with a yellow triangle attached to the left node and a blue rectangle attached to the right node?
The tree has a yellow circle with a blue rectangle attached to the left node. For the blue rectangle, the left node is connected to a red triangle, and the right node is connected to a red rectangle. The red rectangle, the right node is connected to the green square.
Then our thinking model would look like the following diagram.
defFind(Tree): match Tree: case Square('green' , Circle('red') as left , _ ) as root: print(f'case1 \n {root = } \n {left = } \n ') case Geometric( _ , Triangle('yellow') as left, Rectangle('blue') as right, ): print(f'case2 \n {left = } \n {right = } \n') case Circle('yellow', _ , Rectangle('blue', Triangle('red'), Rectangle('red', _ , Square('green') ), ) as right ) as root: print(f'case3 \n {root = } \n {right = } \n ') case _: print('Not Found \n')
From this example, we can see the power of pattern matching, not only to compare nodes with explicit conditions, but also to compare fuzzy conditions as in case 2. It’s not just easier to write code, it’s easier to map the ideas we have in our head directly to the code.
In fact, database and big data engine optimizers make heavy use of pattern matching (not in Python) because the patterns that optimizers need to match are so complex that it would be very difficult to maintain code that simply uses if-else.
Conclusion
Pattern matching is a powerful technique, it’s not just a simple if-else or switch-case, it’s a “pattern matching”.
However, in our daily lives, we seldom really utilize the power of pattern matching, both because we seldom deal with a large number of nested structures, and because we misunderstand the scenarios applied to them.
Here’s a classic example from Stack Overflow. I’ve captured the answer with the most likes.
1 2 3 4 5 6 7
match a: case _ if a < 42: print('Less') case _ if a == 42: print('The answer') case _ if a > 42: print('Greater')
Is this really better than if-else?
1 2 3 4 5 6
if a < 42: print('Less') elif a == 42: print('The answer') elif a > 42: print('Greater')
I believe that most people don’t think so, and this is a typical misapplication scenario.
To sum up, the scenario where pattern matching is used is not conditional comparison, but pattern comparison. In addition, the real power of pattern matching is the nested structure.
Empower Sharding Strategy to Handle Cross-Shard Queries with Confidence
Before discussing sharding, let’s first talk about scaling. We all know that there are two types of scaling, one is vertical scaling, also known as scale-up, and the other is horizontal scaling, aka scale-out.
As for scaling, there are different ways for different purposes. For MongoDB, if we want to improve query performance, then vertical scaling is to improve the machine specs, while horizontal scaling is to increase the number of replicas so that the query can be executed on the idle replicas.
On the other hand, to increase the amount of data stored, for MongoDB, vertical scaling is still about improving the specs of the machine (in this case, the size of the hard drives), while horizontal scaling is about sharding, which is the main topic of this article.
Therefore, we have to understand that the purpose of sharding is to make the data evenly distributed so that MongoDB can store more data, not to improve the query performance. In other words, query performance will be improved by sharding as an extra, but not the main purpose.
Why specifically mention cross-shard queries in the title?
When we consider doing sharding, we will always consider carefully how to choose the shard key, choose the shard key and worry it will affect the production of the query performance, and consider rewrite all the query drastically. These concerns are actually unnecessary.
Let me conclude that if migrating to a sharded cluster from no sharding, as long as the shard key is chosen correctly, then it will only be better, not worse, because it’s a WORST case now.
Why so sure?
The formula for query time in database is as follows.
Ttotal = Ts + Tx
Ts is the retrieval time of the database itself, either by index or by scanning the full collection. Tx is the return time of the query result.
In addition, Ts and Tx have the following properties. Ts ∝ total data volume Tx ∝ result size
When we migrate this collection to a sharded cluster, then the above diagram will change a bit.
The formula will also change a little bit, but the principle is the same.
TclusteredTotal = Max( T1s + T1x , T2s + T2x )
Even if there is data skew happening, I believe we all agree the following conditions are true.
Tns < Ts
Tnx < Tx
Therefore, a conclusion can be deduced.
TclusteredTotal < Ttotal
Nevertheless, there are still some worst cases, for example, sharding is based on the data size rather than the frequency of data access, if there is busy collection(s) originally distributed in two shards without sharding, so they can consume all the resources of their respective machines individually. However, because of sharding, most of the data is distributed to the same shard, which competes for resources.
In other words, the precondition for the problem is the original MongoDB is nearly full and resources are almost exhausted.
How to choose shard key?
From the above introduction, we know as long as the shard key is chosen correctly, then we don’t need to worry about the performance of queries decreasing after sharding. Therefore, how to choose the shard key is pretty important.
I have already provided the ideal formula and described the details in my previous article.
Therefore, in this article, I will only outline which three types of bad cases must be avoided.
Low Cardinality
Suppose we choose an enum field as the shard key, and the range is fixed to [1, 2, 3].
Then even if we write 2 to the max, we still can’t start rebalance, and the maximum number of shards is 3.
Ascending
If we use a continuously incrementing field as the shard key, then we can ensure the chunks are evenly distributed, but it will continuously trigger a rebalance, and the performance will be horrible.
Furthermore, the data we need to query frequently is usually new, in other words, the query and the rebalance will often mix together, making the situation even worse.
Random or Hash Value
Another common choice is to use a random field (or hashing) as the shard key, which also ensures the chunks are evenly distributed and does not trigger frequent rebalances, but the rebalance overhead is increased.
MongoDB’s storage engine, WiredTiger, uses a storage distribution similar to MySQL’s InnoDB, i.e., the primary key (_id) is used as the storage unit (block), and similar primary keys are placed in the same block.
If we use random shard key, it will trigger random access instead of sequential access when we do rebalance.
For instance, when a chunk is full and needs to be rebalanced, the data needs to be migrated across three blocks, and the entire rebalance time will be longer. This is far worse than the performance of a single block to complete the rebalance, as shown in the following diagram.
Conclusion
When we encounter a new problem, we often feel hesitant or even scared because it is unfamiliar.
What we need to do is to take a deep breath and return to the nature of the problem, analyze the core of the problem, and don’t be confused by the external appearance. Taking cross-shard queries as an example, when we use the database perspective instead of MongoDB to unpack the problem, the answer will be obvious.
Software engineering is already a stable field, most of the problems are similar in nature, so don’t panic, think carefully, most of the answers are not complicated.
For a public API to be called by a user, it is common practice to use versioning to control the impact. For example, when a user calls an API and expects certain results, a change in the API’s interface or behavior can cause unpredictable risks for the caller. Therefore, in practice, the original API will be tuned to version 1, and the modified API to version 2.
Users can be sure that calling the original version 1 API will not cause any problems, and the service providing the API can continue to iterate on the functionality, both sides of the development cycle can be independent of each other. Of course, maintaining two sets of APIs increases the maintenance effort for the service provider, so APIs have a lifecycle and don’t live forever.
In the case of the AWS services, the APIs called by the SDK have a version number (named by date), so it is recommended to specify the version of the used API in the production environment.
The following is an example of the AWS Python SDK, boto3, which according to the official document, api_versions should be specified in AWS configure. The version change of each service is available in this channel.
In fact, there are many different ways of implementing API versioning, and one interesting example is Shopify, according to Shopify’s official document, we know that Shopify releases one version per quarter and maintains only four versions at a time, i.e. one year.
Versions are date-named and embedded directly in the URL, and those out-of-date versions fallback to the oldest version in available support.
In other words, suppose the current four versions are as follows.
2023–07
2023–04
2023–01
2022–10
Then the call to 2022–07 would use the version 2022–10.
This is an interesting way of implementation, as the client has one year to make changes, and if it doesn’t, it will still work, but with unexpected results, instead of just crashing.
How to implement such a versioning mechanism? This article provides a possible approach.
Implementation approach
The full source is in the following repo.
The overall implementation architecture is as follows.
All three services implement two URIs: /hello and /hi, which just print the URI with a version number.
For users, to call the corresponding service, they just put a prefix in front of the URI, e.g. curl http://localhost/v2/hello would print out
hello v2
In addition, calling a non-white-listed (v1, v2 and v3) version will fallback to v1, e.g., curl http://localhost/v4/hello will print
hello v1
The core of this experiment is nginx on the gateway.
If the location matches the previous 3 rules, then rewrite the original URI, remove the prefix and redirect to the corresponding service, if it doesn’t match the previous rules but matches the version specification (v plus integer), then redirect to the v1 service anyway.
By using nginx regex, we can make the version match the corresponding service and implement the extra fallback mechanism.
Conclusion
Actually, there is another approach to provide various versions of the API in the original service, for instance, opening several endpoints directly in the API service as follows.
/v1/hello
/v2/hello
/v3/hello
Creating three versions of the API directly instead of adding a prefix to the API through the gateway is not recommended. If you don’t physically isolate them, it adds a lot of development overhead, as in the following real-world example.
When we need to modify the behavior of the common lib, it will inevitably affect v1, which makes it difficult to iterate on the functionality, and also introduces risk to the user.
Therefore, isolation at the physical level is more controllable than isolation at the logical level. This article provides a possible approach, but there are many other implementations that can achieve the same result, so feel free to share them with me.
An Open Source, Flexible and Powerful Event Tracking System
SNOWPLOW is a platform for tracking events, i.e. collecting various user behavior events and analyzing them. Nowadays, there are many famous similar products like Google Analytics, Mixpanel, etc., but in the open source ecosystem, SNOWPLOW has a place.
Although there is a paid version of SNOWPLOW, I can’t figure out any reason not to use Mixpanel instead of SNOWPLOW since I have to pay for it, but there are a lot of advantages I can put forward for the open source version of SNOWPLOW.
For example, the architecture is simple and easy to integrate with existing infrastructure, and there are many flexible components that can be customized. Most importantly, there are a lot of SDKs available for various scenarios, e.g., Web, Mobile apps and even backend systems.
This article will not introduce the use case of SNOWPLOW, because it has been introduced in detail in the official document. Instead, this article will introduce the architecture of SNOWPLOW and provide a free playground.
Why emphasize on free?
In fact, in the official SNOWPLOW document there is a quick start environment for the terraform, but it is deployed on AWS or GCP and uses a lot of paid services. For a beta player, maybe we just want to experience what it can do, but don’t want to pay for it, at least I don’t, then a local free test environment is necessary.
Architecture Overview
First of all, let’s take a quick look at the SNOWPLOW architecture.
There are three core components in the whole infrastructure, collector, enricher and schema registry, which are the fundamentals of SNOWPLOW. As for what kind of data warehouse or analysis engine to use, those can be freely matched and are not part of the SNOWPLOW package.
Collector: Collects events from the SDK, these events are raw data and therefore continue to be delivered to the enricher.
Schema Registry: The registry used in SNOWPLOW is a self-developed iglu, which is a service that manages JSON schema.
Enricher: After receiving the raw data, the first step is to validate it with the schema registered in the schema registry, and if it passes the validation, then it will be processed to produce a more analytical format and continue to send it to the next stage. As for the data to be sent to the data warehouse or Looker or other analysis platforms is based on demand.
One of the more interesting parts of the process is the enrichment stage, where here lists all the available plugins. Let’s take a practical example. An incoming event will only have an IP field, but subsequent analysis will depend on the geographic location, which can be enriched by the plugin IP Lookup.
This is a simple but powerful architecture that includes both validation and enrichment, and what’s more, the events are flexible in their format and can be customized to fit every need. In addition, these events can be downstreamed in a variety of ways based on demand, and can be flexibly integrated with the existing infrastructure.
This Github repository provides a docker-compose.yml, yes, we all love docker-compose. All the required components can be built locally and the usage is written in the README which should not cause any problems.
Also, for testing purposes, in addition to viewing the local database, I’ve put an additional Kafka management console at localhost:9021, which makes it possible to visually see each event.
All the settings of the components to be modified are located in the config.hocon under the corresponding folder.
Let’s describe the architecture of the test environment as follows.
There is a mock web with Javascript SDK installed, after entering http://localhost it will send some events to the collector periodically.
When the collector receives an event, it will store it in the database (atomic.events) and send it to Kafka.
When the Enricher receives an event, it first confirms the schema with the Iglu server, then performs basic enrichment and sends it to Kafka.
The final mock processor is to simulate how the enriched event should be handled.
The whole process is straightforward. After setting up the environment, we can observe the complete data flow in Kafka’s management console.
Conclusion
In fact, I have tried SNOWPLOW’s cloud-based Enterprise version, and the biggest difference I felt was the schema management. In the the open-source version, if we want to customize the event fields, we have to integrate Iglu’s REST API, and then we have to understand Iglu’s design concepts. However, in the cloud version, there is an easy-to-use UI that makes managing schema much easier.
Nevertheless, if we just want to do basic event tracking, I believe the open-source version of SNOWPLOW provides a good capability. Of course, if we want to use it more deeply, we need to understand and integrate it more comprehensively, which is also the price to pay for open source software.
In the previous article, we introduced three kinds of optimization mechanisms for Flink SQL as follows.
Reduce sub plan
Mini batch
Local-Global aggregation
These mechanisms correspond to some use cases individually. Among them, mini batch and Local-Global aggregation are both optimized for GROUP BY operations. However, in the previous article, we mentioned that even though both mechanisms can improve the performance of GROUP BY, they are not applicable to DISTINCT.
Therefore, in these articles, we will start by explaining the reason for this problem, and then we will introduce more optimization mechanisms.
Before explaining the problems encountered by DISTINCT, let’s review Local-Global aggregation with an example.
1 2 3
SELECT color, COUNT(DISTINCT id) FROM T GROUPBY color
This SQL command is a little different from the previous one, i.e., it uses DISTINCT, but it is similar to the previous one.
In Local-Global aggregation, we will do a first aggregation in the mini batch according to the color, and then we will do a second aggregation in the next operator with the results of the mini batch pre-aggregation.
As we can see above, even though we did the pre-aggregation in the mini batch, the effect is not significant because the operation DISTINCT is not able to merge. This also results in a lot of data in the final aggregation, and the data skew is not solved.
Since so, can we use id to group the data again during local aggregation? By splitting again, we can collect data with the same id together, and then we can merge them.
This is exactly the concept of Split Distinct aggregation.
Let’s use pseudocode to explain.
1 2 3 4 5 6 7
SELECT color, SUM(cnt) FROM ( SELECT color, COUNT(DISTINCT id) as cnt FROM T GROUPBY color, MOD(HASH_CODE(id), 4) ) GROUPBY color
Split Distinct aggregation rewrites the original COUNT(DISTINCT id) into the above code. By first splitting the group with id, in this case into 4 groups, the pre-aggregation can be done.
Therefore, the actual operator will look like the following diagram.
We can see in the final aggregation stage that each operator needs to process the data more evenly, in other words, the data skew is solved.
There are two settings to enable Split Distinct aggregation.
table.optimizer.distinct-agg.split.enabled
table.optimizer.distinct-agg.split.bucket-num
The first setting is the feature toggle, and the second setting is the number of groups. Although we use COUNT as an example, any operation that is able to be merged can be split.
Nevertheless, one of the problems with Split Distinct aggregation is the significantly larger state and the increased state access. This is because the result of the split operation relies on the state to be persistent.
How to solve it?
Well, it’s as simple as a Local-Global aggregation after splitting the group. So the complete settings are as follows.
table.exec.mini-batch.enabled
table.exec.mini-batch.allow-latency
table.exec.mini-batch.size
table.optimizer.agg-phase-strategy: “TWO_PHASE”
table.optimizer.distinct-agg.split.enabled
table.optimizer.distinct-agg.split.bucket-num
We have turned on all the optimization settings related to GROUP BY. Be sure to remember that these come at a price, the most obvious of them is the increased use of computing resources.
JOIN Optimization
JOIN is a common practice for enrichment. In a normal SQL JOIN, it’s to find the same columns in the left and right tables and join the remaining columns together, but in streaming, it’s far from simple.
Because, there is no physical table in streaming.
So, in order to make the table concept, Flink will store the data in the state, and when there is an event input, the corresponding data can be taken out from the state immediately. However, there is a big problem with this approach, that is, the state will grow infinitely.
For example, suppose Flink has two streams, order and product. The order stream will keep append once there is a new order, and the product stream will have events for product creation, update and deletion.
Therefore, we write the following Flink SQL.
1 2 3
SELECT*FROM Orders INNERJOIN Product ON Orders.productId = Product.id
In the implementation behind Flink, all changes to the product are stored by the state, so that the corresponding result can be found as soon as the order event is generated, even if the product does not have any orders at all.
When there are a lot of products and the operator is running for a long time, this state can become very huge.
The most straightforward solution to reduce the unused state is to set TTL as follows.
table.exec.state.ttl
Nevertheless, when a product’s state is deleted, it can lead to unexpected results. In the above example, when the state of a product is deleted, if the order event of the same product comes in, the corresponding product information will not be found.
In addition to this simple and brutal approach, Flink also provides three alternatives to JOIN optimization.
It has a similar purpose to setting the TTL directly above, except that it allows the user to decide which field to use as the basis for time judgments.
1 2 3 4
SELECT* FROM Orders o, Shipments s WHERE o.id = s.order_id AND o.order_time BETWEEN s.ship_time -INTERVAL'4'HOURAND s.ship_time
Through this SQL, we can specify order_time and shipment_time as the basis of TTL, then Flink can know not to keep the state of the difference between order_time and shipment_time for more than 4 hours.
However, there is a limitation in this approach, in addition to the fact table on the left is append only, the dimension table on the right must also be append only, otherwise the time interval will be inconsistent.
Before we explain Temporal Join, let’s look at another example.
1 2 3
SELECT*FROM orders LEFTJOIN currency_rates ON orders.currency = currency_rates.currency
Assuming this is an international order, we need to know what the currency rate is for the order in order to calculate the sales performance of the product.
In order to keep the currency rates available at all times, Flink keeps a complete history of the currency rates. For example.
Therefore, the correct result can be obtained regardless of whether the order is placed at 12:00 or 12:01. However, we don’t really need this historical data, because we only need the current currency of the order, and the past history can be cleared out.
This is the concept of Temporal Join, only keep a latest mapping.
currency: conversion_rate
How to use this approach?
1 2 3
SELECT*FROM orders LEFTJOIN currency_rates FORSYSTEM_TIMEASOF orders.order_time ON orders.currency = currency_rates.currency;
One of the most special is FOR SYSTEM_TIME AS OF, this usage can be referred to SQL:2011 standard, this article will not dive into it.
The concept of Lookup Join is even more simple, since the state of Flink is so big and difficult to manage, it is better to go directly to the external data source, such as MySQL, for every event at the moment.
Time for space.
The entire syntax is written in the same way as the previous Temporal Join, using FOR SYSTEM_TIME AS OF, the only thing worth noting is the usage of proc_time.
1 2 3 4
SELECT o.order_id, o.total, c.country, c.zip FROM Orders AS o JOIN Customers FORSYSTEM_TIMEASOF o.proc_time AS c ON o.customer_id = c.id;
Because it is accessing external storage, Flink also provides some built-in optimization features.
In general, the process of accessing external storage is synchronous, sending a request and waiting for a response before moving on to the next one. However, Flink offers to send all requests at once and then wait for responses asynchronously, and it is turned on by default.
If we want to adjust it manually, we can refer to the Hints mentioned in the official document.
In addition, even if the external storage is accessed asynchronously, the actual operation will wait for the correct order of the responses after receiving them. If the order of the responses is not that important, then we can tell Flink not to wait for the responses to be in order.
Use the SQL command mentioned earlier as an example.
1 2 3 4 5
SELECT/*+ LOOKUP('table'='Customers', 'async'='true'. 'output-mode'='allow_unordered', 'capacity'='100', 'timeout'='180s') */ o.order_id, o.total, c.country, c.zip FROM Orders AS o JOIN Customers FORSYSTEM_TIMEASOF o.proc_time AS c ON o.customer_id = c.id;
It is most intuitive to tell the Flink optimizer what to do through Hints, so that it can be adjusted according to each command, but it is also possible to enable the global setting directly.
table.exec.async-lookup.output-mode
table.exec.async-lookup.buffer-capacity
table.exec.async-lookup.timeout
Mode and timeout are considered simple, as for capacity refers to how many IO command will trigger the actual JOIN operation. But I feel this is a bit abstract, set how much is appropriate is difficult to tell, it still needs to rely on experiments.
The last one is Window Join, which is similar to the concept of Interval Join mentioned earlier, and only retains a specific time range of states instead of all historical states. However, the window mechanism is very complicated and not required, so we will directly attach the example of the official document here without further explanation.
1 2 3 4 5 6 7 8 9 10 11 12 13 14
SELECT L.num as L_Num, L.id as L_Id, R.num as R_Num, R.id as R_Id, COALESCE(L.window_start, R.window_start) as window_start, COALESCE(L.window_end, R.window_end) as window_end FROM ( SELECT*FROMTABLE( TUMBLE(TABLE LeftTable, DESCRIPTOR(row_time), INTERVAL'5' MINUTES) ) ) L FULLJOIN ( SELECT*FROMTABLE( TUMBLE(TABLE RightTable, DESCRIPTOR(row_time), INTERVAL'5' MINUTES) ) ) R ON L.num = R.num AND L.window_start = R.window_start AND L.window_end = R.window_end;
To sum up, all kinds of JOIN rewriting are aimed at reducing the size of state.
The original JOIN will generate extremely large state, which will consume a lot of hardware resources if stored in memory, or generate a lot of hard disk IO if stored on persistent storage such as RocksDB, either of which will have a huge impact on performance.
Therefore, these four JOIN optimizations are all aimed at reducing the size of the state, but the exact solution to be used depends on the use case. The following is a brief list of the applicable scenarios.
Interval Join: Both the fact table and dimension table are append-only.
Temporal Join: Only the latest dimension is required.
Lookup Join: The dimension table is stored externally and does not care about dimension changes.
Window Join: Both fact and dimension tables have the window function enabled.
Conclusion
In the previous article, we introduced how to optimize general SQL commands and how to make GROUP BY more efficient. In this article, we introduced DISTINCT and JOIN.
I believe these two articles should have covered most of the Flink SQL scenarios, in fact, it is not hard to find that there are the following ways to make Flink SQL perform better.
Reducing invalid ( repeated ) commands
Reducing state access
Reducing the size of state
Reducing data skew
Every optimization has a proper scenario and a price to pay, and how to make Flink SQL perform better is a balance between these tradeoffs. Most of the optimizations are case-by-case and require understanding of Flink implementation before they can be applied.
These two articles are based on the latest stable version of Flink 1.17, maybe the URL of the attached reference will change, but I have explained the core concepts, so the new or old version of Flink should also be used as a reference.
Before talking about the tuning, we have to understand how Flink SQL works. In general, there are several steps as follows.
After receiving a SQL command, it is first parsed into a logical plan, and then the execution plan is generated by the optimizer. With the execution plan, the actual code for execution can be generated and finally turned into a job.
The only step we can control on the whole process is the optimizer.
And, the following are factors affecting Optimizer.
LogicalPlan: Generated from SQL parser, it can be said that a good writing style basically solves most of the performance problems.
FlinkConf: In Flink, there are many optimizer-related settings that can be adjusted.
SchemaDesign: Of course, the schema design also determines some of the performance, such as the primary key.
Hints: Although the optimizer will try to cover as many scenarios as possible, sometimes we still need to “tell” the optimizer what to do, and that can be done through hints.
Thus, this article will introduce a few common scenarios on how to tune.
Reuse Sub Plan
In Flink’s settings, table.optimizer.reuse-sub-plan-enabled is turned on by default. The functionality of this setting can be explained by a simple example.
1 2
INSERTINTO sink1 SELECT*FROMtableWHERE a >0; INSERTINTO sink2 SELECT*FROMtableWHERE a >0AND b <0;
First, convert the above two SQL commands into a logical plan.
Then, after the reuse-sub-plan of the optimizer, the following execution plan will be generated.
What the optimizer does is to merge the same parts, so that there is no need to repeat the process. But from our naked eyes, we can still see one identical part can be merged.
WHERE a > 0
However, the optimizer does not recognize this clause, so we need to rewrite the original SQL to make the optimizer understand the merging rules for this filter.
1 2 3
CREATE TEMPORARY VIEW v SELECT*FROMtableWHERE a >0; INSERTINTO sink1 SELECT*FROM v; INSERTINTO sink2 SELECT*FROM v WHERE b <0;
By creating a temporary view, the optimizer can generate a more efficient execution plan.
Mini batch is a built-in performance optimization mechanism for Flink. The mechanism behind is explained in the official document.
The above diagram shows an overview of a mini batch, in plain English, mini batch drastically reduces the number of accesses to the state.
Let’s use a practical example to explain this.
1 2 3
SELECT color, SUM(id) FROM T GROUPBY color
This is a typical SQL aggregation operation, in fact, the internal operation of Flink is each event must access the state once, so it can be summed up.
It is written in Python as pseudocode as follows.
1 2 3 4 5 6
defprocess(input): color, id = extract(input) ret = state.get(color) ret += id state.update(color, ret) yield ret
When inputting an event, first of all we find its color, then we take out the previous summation result, then we add the id to it, and finally we write the result back to the state. The next event comes in and repeats the process.
Such a iteration will be extremely frequent to access the state, resulting in performance impact. So mini batch collects events and accesses the state only once for each specific color.
1 2 3 4 5 6 7
defmini_batch_process(batch): for color in batch: ret = state.get(color) foridin batch[color]: ret += id state.update(color, ret) yield ret
As you can see from the code above, the process of mini batch is to collect the events of the same color, so the sum of the same color can be done at once, without the need to access the state frequently.
To enable mini batch is as simple as turning on the settings of the optimizer.
table.exec.mini-batch.enabled
table.exec.mini-batch.allow-latency
table.exec.mini-batch.size
The first setting is on/off, the rest is to determine the size of the batch, which can be based on the amount or duration.
There is a further optimization approach to GROUP BY, called Local-Global aggregation, which is an extension of mini batch.
The biggest problem of mini batch is that it will be grouped by color. If a particular color is especially large, then the downstream operator assigned to this color will have to process much more data, which is also called data skew. Therefore, Flink provides a mechanism to perform pre-aggregation in the mini batch in order to reduce the downstream load.
From the above diagram, we can see that after turning on Local-Global aggregation, the summation operation will be done inside the mini batch, and the last data aggregation will be done with a small amount of data. In the example red Agg, the amount of processed data will be changed from 12 to 3, i.e., the data skewing problem is solved.
To enable Local-Global aggregation, it is necessary to use the following settings in addition to the mini batch already enabled.
table.optimizer.agg-phase-strategy: “TWO_PHASE”
Conclusion
This article first introduces three of Flink’s built-in optimization mechanisms:
Reduce sub plan
MiniBatch aggregation
Local-Global aggregation
Only the first one is enabled by default, but there is still room for manual optimization. The other two are turned off by default, especially the Local-Global aggregation, which requires an additional mini batch to be enabled in addition to the two phases.
Since these mechanisms look so good, why are they not turned on by default?
Because, these mechanisms are not without cost.
In the case of mini batch, when mini batch is turned on, it means the freshness of data will be affected, depending on the mini batch settings.
In addition, Local-Global aggregation is not available for all operations. It is only used for aggregations that can be merged, such as SUM, MAX, COUNT, etc., and not for DISTINCT. Moreover, Local-Global aggregation increases the computation complexity and therefore consumes additional computing resources, which is not always reasonable if the data skew is not large.
These optimization mechanisms must be understood before using in order to use them correctly, and turning them all on unconditionally may lead to counterproductive results.
Of course, Flink’s optimization mechanism is not just these, next time we will talk about DISTINCT and JOIN which are not covered in this article.
Explaining how to use ElasticSearch in a better way
Last time, we have introduced some tips for boosting ElasticSearch performance.
In addition, that article explains the underlying details of ElasticSearch. This time, we are going to talk a little more about best practices for using ElasticSearch.
These practices are general recommendations and can be applied to any use cases. Let’s go.
Bulk Requests: The Bulk API makes it possible to perform many index/delete operations in a single API call. This can greatly increase the indexing speed. Each subrequest is executed independently, so the failure of one subrequest won’t affect the success of the others. If any of the requests fail, the top-level error flag is set to true and the error details will be reported under the relevant request.
Multithread clients to Index Data: A single thread sending bulk requests is unlikely to be able to max out the indexing capacity of an ElasticSearch cluster. In order to use all resources of the cluster, you should send data from multiple threads or processes. In addition to making better use of the resources of the cluster, this should help reduce the cost of each fsync. Both the index data and transaction log are periodically flushed to disk. If there are more data with multithread, the more data is synced to disk to reduce I/O to improve performance.
index.refresh_interval: By default, ElasticSearch periodically refreshes indices every second, but only on indices that have received one search request or more in the last 30 seconds.This is the optimal configuration if you have no or very little search traffic (e.g. less than one search request every 5 minutes) and want to optimize for indexing speed. This behavior aims to automatically optimize bulk indexing in the default case when no searches are performed. In order to opt out of this behavior set the refresh interval explicitly. On the other hand, if your index experiences regular search requests, this default behavior means that ElasticSearch will refresh your index every 1 second. If you can afford to increase the amount of time between when a document gets indexed and when it becomes visible, increasing the index.refresh_interval to a larger value, e.g. 30s, might help improve indexing speed.
Auto generated IDs: When indexing a document that has an explicit id, ElasticSearch needs to check whether a document with the same id already exists within the same shard, which is a costly operation and gets even more costly as the index grows. By using auto- generated ids, ElasticSearch can skip this check, which makes indexing faster.
index.translog.sync_interval: This parameter determines how often the translog is fsynced to disk and committed, regardless of write operations. Defaults to 5s. Values less than 100ms are not allowed.
index.translog.flush_threshold_size: The translog stores all operations that are not yet safely persisted in Lucene (i.e., are not part of a Lucene commit point). Although these operations are available for reads, they will need to be replayed if the shard was stopped and had to be recovered. This setting controls the maximum total size of these operations, to prevent recoveries from taking too long. Once the maximum size has been reached a flush will happen, generating a new Lucene commit point. Defaults to 512mb.
Large Documents: Large documents put more stress on network, memory usage and disk. Indexing large document can use an amount of memory that is a multiplier of the original size of the document. Proximity search (phrase queries for instance) and highlighting also becomes more expensive since their cost directly depends on the size of the original document.
Set Index Mapping Explicitly: ElasticSearch can create mapping dynamically, but it might be not suitable for all scenarios. For example, the default string field mappings in ElasticSearch 5.x are both “keyword” and “text” types. It’s unnecessary in a lot of scenarios.
Index Mapping — Nested Types: Querying on nested fields is slower compared to fields in parent document. Retrieval of matching nested fields adds an additional slowdown. Once you update any field of a document containing nested fields, independent of whether you updated a nested field or not, all the underlying Lucene documents (parent and all its nested children) need to be marked as deleted and rewritten. In addition to slowing down your updates, such an operation also creates garbage to be cleaned up by segment merging later on.
Index Mapping: Disable the _allfield concatenates the values of all other fields into one string. It requires more CPU and disk space than other fields. Most use cases don’t require the _all field. You can concatenate multiple fields using the copy_to parameter. The _all field is disabled by default in ElasticSearch versions 6.0 and later. To disable the _all field in earlier versions, set enabled to false.
Leverage Index Templates: Index templates define settings like number of shards, replicas and mappings that you can automatically apply when creating new indices. ElasticSearch applies templates to new indices based on an index pattern that matches the index name.
Use Replicas for Scalability & Resilience: ElasticSearch is built to be always available and to scale with your needs. It does this by being distributed in nature. You can add nodes to a cluster to increase capacity and ElasticSearch automatically distributes your data and query load across all of the available nodes. For ElasticSearch to be highly available, its indices needs to have fault tolerance in place. This can be achieved using replica shards. A replica shard is a copy of a primary shard. Replicas provide redundant copies of your data to protect against hardware failure and increase capacity to serve read requests like searching or retrieving a document.
Shard Sizing: A shard is a Lucene index under the covers, which uses file handles, memory, and CPU cycles. Default shard strategy for an index in ES is 5 primary shards with a replica. The goal of choosing a number of shards is to distribute an index evenly across all data nodes in the cluster. However, these shards shouldn’t be too large or too numerous. A good rule of thumb is to try to keep shard size between 10–50 GB. Large shards can make it difficult for ElasticSearch to recover from failure, but because each shard uses some amount of CPU and memory, having too many small shards can cause performance issues and out of memory errors.
Keep shard count of index in multiple of data nodes with equivalent size and distributed across nodes. Set a primary shard count, via a template, by targeting 50GB max per primary shard (log analytics) or 30GB max (search use cases) Shard assignment across data nodes will happen based on 2 important rules.
Primary and replica shard of same index will not be assigned on same data nodes.
A shard is placed on a node based on how many shards are available on that node or to equalize the number of shards per index across all nodes in the cluster. Also note there can be chances of bigger shards allocated to some node and smaller to others. It is recommended that shard count of an index (primary + replica) should be multiple of data node count. Let’s say, you have a 4 node cluster, the total shards (primary + Replica) for your index should be either 4 or 8 or 12 etc. This ensures that the data is evenly distributed across the nodes.
Index State Management: ISM lets you define custom management policies to automate routine tasks and apply them to indices and index patterns. You no longer need to set up and manage external processes to run your index operations. A policy contains a default state and a list of states for the index to transition between. Within each state, you can define a list of actions to perform and conditions that trigger these transitions. A typical use case is to periodically delete old indices after a certain period of time.
Organize the data in index by date: For most logging or monitoring use cases, we can organize indices to be daily, weekly, or monthly, and then we can get an index list by a specified date range. ElasticSearch only needs to query on a smaller dataset instead of the whole dataset. In addition, it would be easy to shrink/delete the old indices when data has expired.
Use Curator to rotate data: Curator offers numerous filters to help you identify indices and snapshots that meet certain criteria, such as indices created more than 60 days ago or snapshots that failed to complete
