Log processing is a broad field that includes real-time computing, data warehousing, and offline computing. This topic describes how to process logs in order, without data loss, and without duplication in real-time computing scenarios. It also explains how to maintain these properties when upstream and downstream systems are unreliable or service traffic fluctuates.
To make the concepts easy to understand, this topic uses "A Day at the Bank" as an analogy. This topic concludes by describing the consumer group feature of Simple Log Service LogStore and explains how to use this feature with tools such as Spark Streaming and Storm Spout to process log data.
What data can be abstracted as logs?
Half a century ago, logs were thick notebooks kept by ship captains and operators. Today, computers generate and consume logs everywhere. Servers, routers, sensors, GPS, orders, and various devices describe our world from different angles. A captain's log, in addition to a timestamp, could contain almost any content, such as a text record, a picture, weather conditions, or the ship's direction. Today, this "captain's log" approach has expanded to many areas, representing an order, a payment record, a user visit, or a database operation.
In computing, common logs include metrics, binary logging (Binlog) for databases and NoSQL systems, events, auditing logs, and access logs.
In the analogy used in this topic, a user's operation at a bank is treated as a single log entry. This entry includes the user's name, account name, operation time, operation type, and transaction amount.
For example:
2016-06-28 08:00:00 Zhang San Deposit 1000 USD
2016-06-27 09:00:00 Li Si Withdrawal 20000 USDThe LogStore data model
To abstract the problem, this section uses the Simple Log Service LogStore as a model. The model includes the following components:
Log: A record that consists of a time and a set of key-value pairs.
LogGroup: A collection of logs that share the same metadata, such as IP address and source.
The relationship between these components is as follows:
Shard: A partition that serves as the basic unit for reading and writing a LogGroup. You can think of it as a first in, first out (FIFO) queue with a 48-hour data lifecycle. Each shard provides a write throughput of 5 MB/s and a read throughput of 10 MB/s. A shard has a logical range (BeginKey, EndKey) to group different types of data.
LogStore: A log repository for storing a specific type of log data. A LogStore is a container built from a combination of shards in the
[0000,FFFF..)range and contains one or more shards.Project: A container for LogStores.
The relationships among these concepts are as follows.
A day at the bank
Take a 19th-century bank as an example. In a city, several users (producers) go to the bank to deposit or withdraw money (user operations). The bank has several clerks (consumers). Because computers for real-time synchronization did not exist in the 19th century, each clerk kept a small ledger to record information. Every evening, they took the money and ledgers to the main office for reconciliation.
In a distributed system, a clerk can be compared to a single machine with fixed memory and computing power. Users represent requests from various data sources. The bank hall represents the LogStore that processes user deposit and withdrawal data.
In this analogy, the roles and their main operations include the following:
Log/LogGroup: User operations such as deposits and withdrawals.
User: The producer of logs and log groups.
Clerk: The employee who processes user requests at the bank.
Bank hall (LogStore): Where user requests first arrive before being assigned to a clerk for processing.
Shard: The method by which the bank hall organizes user requests.
Problem 1: Ordering
A bank has two tellers (A and B). Zhang San enters the bank and deposits 1000 CNY with teller A, who records the 1000 CNY deposit in their own ledger. Later that afternoon, Zhang San is short on cash and goes to teller B to withdraw money. When Teller B checks their ledger, they find a problem: Zhang San has not deposited any money there.
This example shows that deposits and withdrawals are strictly ordered operations. The same clerk (processor) must handle all operations for the same user to maintain state consistency.
The method to ensure order is simple: create a queue. You can create one shard and have only one clerk, A, process the requests. User requests are handled on a first in, first out basis, which works perfectly. However, this method is inefficient. If 1,000 users need to perform operations, adding more clerks will not help. What can be done in this scenario?
Suppose you have 10 tellers. You can create 10 shards, one for each teller. To ensure that operations for the same account are processed in order, you can use consistent hashing to map users to specific shards. For example, you can create 10 queues that function as shards and assign each teller to process a single shard. Different bank account numbers or user names are then mapped to a specific shard. For instance, the user name 'Zhang San' is hashed to a value that consistently maps to the same shard. This ensures that all requests from Zhang San are always processed by the same teller, such as Teller A.
Of course, if many users have names that start with the same letter, you can use other strategies. For example, you can hash based on the user's AccountID or ZipCode. This can distribute operation requests more evenly across the shards.
Problem 2: At-least-once processing
Alice is with Clerk A to make a deposit. Halfway through, Clerk A takes a phone call. When they return, they mistakenly think the transaction is complete and start serving the next user. As a result, Alice's deposit request is lost.
Although machines are not prone to human error and have higher uptime and reliability than clerks, they can still experience failures or processing interruptions due to high load. Losing a user's deposit in such a scenario has unacceptable consequences.
Clerk A can record their progress in a personal journal, separate from the official ledger, noting which position in the shard has been processed. Only after Alice's deposit request is fully confirmed can Clerk A call the next person.
What is the problem with this approach? It can lead to duplication. For example, Clerk A has already processed Alice's request and updated the ledger. They are about to record the progress in their journal when they are suddenly called away. When Clerk A returns, they see that Alice's request was not recorded as processed. Clerk A will then process the request again, causing a duplicate transaction.
Problem 3: Exactly once processing
Does duplication always cause problems? Not necessarily.
In idempotent situations, duplication is wasteful but does not affect the final result. An operation is idempotent if repeated consumption does not change the outcome. For example, a "check balance" operation is a read-only operation. Repeating it does not affect the result. For non-read-only operations, such as deregistering a user, repeating the operation does not cause an issue.
However, most real-world operations, such as deposits and withdrawals, are not idempotent. Repeating them can have a significant impact on the results. The solution is for the clerk (A) to combine two actions into a single transaction: completing the ledger entry and marking the processing as complete in the journal. This combined action is recorded as a checkpoint.
If Clerk A leaves temporarily or permanently, other clerks can follow the same rule: if the record shows the operation is complete, they process the next one. If not, they repeat the operation. Atomicity must be guaranteed during this process.
A checkpoint can use the position or time of an element in a shard as a key and store it in a persistent object. This indicates that the current element has been successfully processed.
Business challenges
After explaining these three concepts, the principles may seem simple. However, in the real world, changes and uncertainties in the number of users and processing volume make these problems more complex.
On payday, the number of users increases sharply.
Clerks are not robots. They need vacations and lunch breaks.
To improve the overall service experience, the bank manager needs to add more clerks. What criteria should be used to decide when to add clerks?
During a shift change, can clerks easily hand over their ledgers and records?
A Day in the Life
The bank opens at 8:00 AM.
There is only one shard, Shard0. All user requests are queued in Shard0, and Clerk A can handle them.
Peak hours begin at 10:00 AM.
The bank manager decides to split Shard0 into two new shards, Shard1 and Shard2, after 10:00 AM. The manager sets a rule: users with names starting with letters from A to W queue in Shard1, and users with names starting with
[X, Y, Z]queue in Shard2. The letter ranges for the two shards are uneven because the distribution of user surnames is also uneven. This mapping method ensures that the clerks' workloads are balanced.Request consumption status from 10:00 AM to 12:00 PM.
Clerk A struggles to handle both new shards, so the manager brings in Clerks B and C. Because there are only two shards, Clerk B takes over one shard from Clerk A, and Clerk C remains idle.
More and more people arrive at 12:00 PM.
The bank manager feels that Clerk A is under too much pressure with Shard1. Therefore, the manager splits Shard1 into two new shards, Shard3 and Shard4. Clerk A handles Shard3, and Clerk C handles Shard4. After 12:00 PM, requests that were originally routed to Shard1 are now directed to Shard3 and Shard4.
Request consumption status after 12:00 PM.
The traffic continues until 4:00 PM, then begins to decrease.
Therefore, the bank manager lets Clerks A and B take a break and asks Clerk C to handle the requests in Shard2, Shard3, and Shard4. Gradually, Shard2 and Shard3 are merged into Shard5. Finally, Shard5 and Shard4 are merged into a single shard. The bank closes after all requests in the final shard are processed.
Real-world log processing
The process described above can be abstracted into a classic log processing scenario. To meet the bank's business needs, you need a flexible and elastic basic log framework. This framework should include the following features:
Elastic scaling for shards.
Automatic rebalancing when consumers go online or offline, without data loss. Ordering must be maintained throughout the process.
Exactly-once processing, which requires consumer cooperation.
The ability to monitor consumption progress to properly allocate computing resources.
Support for log ingestion from multiple channels. For a bank, this means opening channels such as online banking, mobile banking, and checks to accept more user requests.
You can use LogStore consumer groups to solve these classic problems in real-time log processing. This lets you focus on your business logic without worrying about traffic scaling, failover, or other details. For more information, see Consume logs using a consumer group.
Additionally, Spark Streaming provides interfaces that are implemented with consumer groups. For more information, see Consume data with Spark Streaming.