Introduction This is a study notes. A study of material from Jay Kreps speaking Log Bowen. Original very long, but I insist on reading, and learned a lot, but also deeply brother to Jay technical capabilities, infrastructure capacity and for a deep understanding of distributed systems impressed. But also because some understanding and views coincide brother Jay and slightly complacent. Jay Kreps Linkedin's former Principal Staff Engineer, lead author of the company's co-founder and current Confluent and CEO, Kafka and Samza of. The so-called notes, just read the article, pick up a pen to remember, because the good brother Jay chapter organization itself, and its own philosophy of science literacy and literacy is also high, so private that the color of things not omitted. I. Sources The Log: What every software engineer should know about real-time data's unifying abstraction Second, note the value of 2.1 Log 1) Log is the core of the following systems: distributed map database distributed search engine and the first generation of Hadoop KV II database 2) Log may be followed as long as the history 2012 Cheap Nike Air Jordan 1 High Heels For Sale Beige Pink of the computer, and is the core of distributed data systems and real-time computing systems. 3) Log name many: Commit logTransaction logWrite-ahead log4) do not understand the Nike Shoes Log, you can not fully understand the database NoSQL storage KV storage replication Paxos algorithm HadoopVersion Control or any software system 2.2 What is a Log? 2.2.1 Overview of record will be appended to the end of the log. From left to right to read records. Each entry has a unique and ordered the log entry number. Order records defines such a concept: time. Because the more records left sooner. Entry of the serial number can be used as a time stamp, the order of the records as the time of this concept may seem strange, but soon you will find that doing so: can easily be 'time' with any particular physical Decoupling clock. Log and common file, table (table) is not so big difference. File is a set of bytes table is a set of records Log can be said for some of the chronological record file or table so that you may feel log so simple, there is a need to discuss it? In fact, the core meaning that log: Log record what happens when (they record what happened and when.). And this one, is usually the most distributed systems the core of things. Note that there is a need to clarify a few concepts: application logs (application logs) Log and programmers usually contact 本篇 discussed different application logs usually a unstructured, record the error message, debugging information for tracking , posters logging applications running, such as to write to the log file by the local log4j or syslog. The log 本篇 in question is accessed programmatically, not posters, such as 'journal', 'data logs'. Application Log is log in 本篇 discussed a specialized. 2.2.2 database LogsLog origins unknown, just like the invention binary search of people are unaware of this invention is an invention. Log in as early appearance and IBM's System R. In the database, the database needs to collapse, holding a variety of data structures and indexes in sync. To ensure atomicity and durability, the database required prior to modify data structures and indexes submit records to modify its content. So log records what happened when, and every table and the index itself, all this historical information mapping. Because the log is immediately persistent, so when the crash occurred, it has become a reliable source of recovery other persistent structures. Log guarantee ACID properties from one implementation, data replication between databases developed into a resort. Obviously, a series of data are changed in the database, the database becomes synchronized between the most needed information. Oracle, MySQL, PostgreSQL, contains the log transfer protocol to send a portion of the log for holding copied from the database (Slave). Oracle's XStreams and GoldenState, the log data subscriptions as a common mechanism, to be supplied to a non-Oracle database subscription data. MySQL and PostgreSQL also provides similar components, which are the core data system architecture. Machine-oriented Log, not only can be used in the database can also be used in: Messaging data stream (data flow) 2.2.3 distributed real-time computing systems logsLog address two very important distributed data systems Problems: 1) Nike Shoes Global data changes ordered 2) distributed data replication of the so-called state machine principle (State Machine Replication Principle): the process if the two determined starting from the same state, the same order, receive the same input, then they will produce the same output, and end in the same state. The so-called defined (deterministic), refers to the process is time-independent, the results are also not affected by the additional input. Examples of non-deterministic through to understand: the order of execution of multiple threads lead to different results in different execution getTimeOfDay () method does not duplicate other so-called status of the process, can be any data on the machine, whether in the end it is the the machine's memory or on the disk. The same input in the same order, produce the same results, it is worth your attention, which is why the log is so important, this is an intuitive concept: If you enter the same log two deterministic program They will produce the same output. In building distributed systems, aware of this, you can make: Let all the machines to do the same thing, the statute is: build distributed, meet consistent log system, that all processing system provides input. Role Log system is uncertainty over all input streams to disperse, to ensure that all processing the same input node replication sync. The most wonderful of this approach is that you can log timestamps index, as all copies of the clock to treat the nodes: node by copying treated largest log timestamps, as the copy node unique ID, Thus, the timestamp combined log, you can uniquely express the entire state of this node. Application of this method is the 2013 Air Jordan 4 (IV) Retro way a lot: Record a request for a service before and after the reply to the request, recorded a series of conversion command changes the service status, or services performed, etc. in the log. Theoretically speaking, we can record a series of machine instructions, or the name and parameters of the method called by the same process as long as the behavior of data Nike shoes online processing, these processes can ensure consistency across nodes. Who often play database, the logical logs and physical log will be treated differently: the physical log: records all the contents of the line change. Logical log: No record changes in the content, but Insert, update, delete lines as a result of changes in the SQL statement. Distributed systems, there are usually two ways to Air Jordan 10 handle replication and data processing: 1) State machine model (active - active) 2) Primary-back model (active - passive) as shown below: In order to understand the two ways different, look at a simple example: Now, clusters need to provide a simple service, to do the addition, multiplication and other arithmetic. Initially, maintain a number, such as 0. Active - active: in some operations such as logging, such as '+1', '* 2' and so on, so that each copy nodes need to perform these operations, in order to ensure that the final status of the data is consistent. Active - passive: a single master node, perform the '+1', '* 2' and other operations, and record the results of operations in the log, such as '1', '3', '6' and so on. The above example also reveals why the order is a key factor to maintain consistency between the replicate sites, if disrupted the order of these operations, you will get a different result of the operation. Distributed log, some consensus algorithm can be used as a data structure: PaxosZABRAFTViewstamped Replication a log, characterized a series of decisions about what is the next value. 2.2.4 Changelog from a database perspective, a set of records of data changes changelog and tables are dual and interoperability. 1) The recording of the data changes log, can be reconstructed in a state table (that may be a non-relational storage systems have key record) 2) On the contrary, if the table is changed, the change can be included in log. This is exactly what you want where quasi-real-time replication Cheats! This and version control done very similar things: management of distributed, concurrent, changes to the state. Version control tools, maintenance reflect the modified patch, which is actually log, you and a checked out (checked out) branch snapshots interact this snapshot is equivalent to a database table. You will find, version control and distributed systems, replication is based on log: when you later, you just pull the reflecting patch version changes, and applied to a snapshot of the current branch. 2.3 Data Integration (Data integration) 2.3.1 Data Integration meaning so-called data integration, is an organization of the data for all services and systems become available. In fact, the effective use of the data, it is consistent with the theory of Maslow's hierarchy of needs. The bottom of the pyramid, is to collect data, which is integrated into the application system (either real-time calculation engine, or text file, or python script). These data, need to go through the conversion, to maintain a unified, standardized, clean format to be easily read and processed. When the above requirements are met, you can begin to consider a variety of data processing methods, such as map - reduce or real-time query system. Obviously, if there is no reliable, comprehensive data stream, Hadoop will just be an expensive, difficult to integrate heater (clusters lot of power it?). On the contrary, if we can ensure that the data stream, reliable, available and complete, we can consider the more advanced gameplay, better data model and consistent, more easily understood semantics. Then, attention can shift to visualization, reporting, and forecasting algorithms up (digging depth ah ah ah machine). 2.3.2 Data integration of two complex event event data recorded events is how it happened, and not just what happened, as this type of log is usually application logs, because normally written by the application system. But this is, in fact, confused the log function. Google's wealth, in fact, is built on a (user) click stream and dislikes impression (experience) over the relevance of pipeline generated, and click stream and the impression that event. A variety of reasons for the outbreak of professional data system these systems exist: online analytical (OLAP) search simple online storage batch profiling, etc. (such as spark) Obviously, you want the data integrated into such a system, for data Integrated speaking, extremely difficult. 2.3.3 based on the data for each source stream on a log-structured logical sense, can be modeled based on log. The data source can be recorded events (clicks and PV) applications, can be acceptable to change the database tables. Log each subscriber, they are as fast as possible from these data sources in acquiring new record, apply to the local storage system, and to enhance its read offset log in (offset). Subscribers can be any data system, such as caching, Hadoop, a database to another site, or search engine. Log, actually provides a logical clock, for data changes, which the state can be measured in different subscribers, because these subscribers log in to read the offset in different and independent of each other, this shift as a 'Time' as the time sense. Consider an example, a database, and some of the cache server: Log provides such a capability, you can make all the cache servers synchronized, and which introduced them 'moment.' Suppose we want to write a number of X's log, a caching server reads from the data, in order not to read the old data, you only need to ensure that: The data in the cache server (synchronous) replication before X this position, we do not It can read anything from this cache. In addition, log buffer also provides the ability to act, in order to support the behavior of producers and consumers in an asynchronous manner. The key reason that supports asynchronous, is to subscribe to the system may crash, offline for maintenance, then resume on the line, but in this case, each subscriber in its own pace of consumer data. A batch system, such as Hadoop, or a data warehouse, based on hours or days of consumption data, and a real-time system, usually consumption data in seconds. The data sources or log, consumption data for subscribers ignorant, therefore, need to do a seamless add subscribers and removing subscribers in the pipeline. More importantly, subscribers only need to know the log, without any understanding of their sources of consumer data, regardless of the data source is an RDBMS, Hadoop, or one of the latest popular KV database, and so on. The reason why the discussion log, rather than the message system, because different messaging systems that ensure different characteristics, and use the messaging system of the word, it is difficult to express a comprehensive and precise semantics, because the message system, and more importantly that the redirect message. However, it can log understood as a messaging system, which provides strong durability to ensure the orderly and semantics, in a communication system, which is called atomic broadcast. 2.4 LinkedinLinkedin current primary system include (Note: 2013): SearchSocial GraphVoldemort (KV storage) Espresso (document storage) Recommendation engineOLAP query engineHadoopTerradataIngraphs (monitor patterns and metrics service) for each system, provides specialized in their areas of expertise Advanced features. (This is too long a period, Jay brother very Nengkan ah, so pick emphasis to remember it!) 1) The reason for introducing the concept of a data stream, is due to the above table oracle database, the establishment of an abstract caching layer for the search engine's index construction and social profiles updated to provide the ability to expand. 2) In order to better deal with some linkedin recommendation algorithm, began to take Hadoop cluster, but the team's experience in this block is still shallow, so take a lot of detours. 3) at the beginning, simply rudely that as long as the data is pulled out from the oracle data warehouse, thrown into hadoop it. The results showed that: first, the data is rapidly exported from the oracle data warehouse is a nightmare; the second is a little worse, some of the data processing in the data warehouse wrong, resulting in a batch job hadoop not result as expected output, and By hadoop batch tasks, generally irreversible, especially in Air Max for Men a later report. 4) Finally, the team abandoned the data from the data warehouse approach, directly to the database and logs for the data source. Next, create a wheel: KV storage (Voldemort). 5) copies of data so that even if the job is not on the tall, also occupy a lot of time to deal with the team, make matters worse, once the data processing pipeline has a point of error, hadoop immediately becomes waste wood, because then cattle force algorithm running on the wrong data, there is only one consequence of this is to produce more incorrect data. 6) even at high levels of abstraction team building thing for each data source or need a specific configuration, which is also the root of many mistakes and failures. 7) a large number of programmers like to follow up, every programmer has a large number of ideas, integrated system, add this functionality, integrate this feature, or if you want to customize the data source. 8) Jay brother began to realize: First, although they are still very rough to build pipelines, but extremely valuable. Even solve the data in the new system (such as hadoop) available issues, but also unlock a large number of possibilities. Previous calculations difficult start possible. New products and analysis, only need to unlock data from other systems, and to integrate, they can easily do it. Second, it is clear, reliable data load requires more solid support, if we can capture all of the structure, it can make hadoop fully automated data loading, no need to add new data sources or manual mode to modify data. Data will magically appear in HDFS, and the new data source is added, Hive table will use the appropriate column automated, adaptive generation. Third, the data coverage inadequate. Because to deal with a lot of new data sources, it is difficult. 9) In order to solve the problem after the data load new data source is added, the team began this attempt: Soon, they found that this does not work out, because the publish and subscribe, production and consumption, or two-way data flow normally, it became an O (n ^ 2) problem. So, they need such a model: the need to isolate from the data source to each consumer, the ideal case, only those consumers to interact with a data repository, and this repository can provide them access to any data. 10) the message system + log = Kafka, kafka turned out. Relationship between 2.5 Log and ETL, data warehouse data warehouse 2.5.1 1) a fitted clean, structured, integrated data repository, for analysis. 2) While the idea is nice, but a little way to get data out of date: periodically acquires data from the database, convert it to something better readable format. Data Warehouse Question 3) before that: the clean data and data warehousing highly coupled. Data warehouse, it should be a collection of query functions, these functions serve to reports, search, ad hot analyzes, including the counting (counting), aggregation (aggregation), filtration (filtering) operation, so it should be a batch system. But the clean data and such a batch system is highly coupled together, means that these data can not be consumed in real-time systems, such as search engine's index construction, real-time computing and real-time monitoring system, and so on. 2.5.2 ETLJay brother believes, ETL nothing more than to do two things: 1) data extraction and cleaning, unlock the data from a particular system 2) to reconstruct the data, it can query the data warehouse. For example, the data type to the type of adaptation to a relational database, the mode is converted to a star or snowflake schema, or break it down into some kind of column-oriented storage format. However, these two things are coupled together, a big problem, because the integrated after the clean data, this should be the other real-time systems, the index build system, low latency processing system consumption. Data warehouse team, responsible for collecting and cleaning data, however, the producers of Air Jordan 11 these data are often unclear because the data processing needs of data warehouse team, resulting output is difficult to be extracted and cleaned data. Also, because the core business team and the company to keep pace with other teams this child is not sensitive, so the data can be processed real coverage is low, the data flow is very fragile and difficult to quickly respond to change. Therefore, the better way is: If you want a clean data set to do the search, real-time monitoring trends, real-time alerts thing to the original data warehouse or hadoop cluster as infrastructure, are inappropriate a. Worse, the built ETL for data warehouse loading systems for other (real) system with no little. The best model, that is, before the data released by publishing data, it has completed the cleaning process the data, because only the publishers they know best what kind of data. And all operations done at this stage, should meet lossless and reversible. All the rich semantics, or real-time conversion add value should be treated (post-processing), including the establishment of the log after the original release for the event data session, or add certain field of interest. The original log still can be used alone, but also real-time applications such derive new parameters of log. Finally, only correspond to a specific target system data aggregation operation, it should be used as part of the data load, such as transforming to a star or snowflake schema, for analysis in the data warehouse and the reports. Because at this stage, like a traditional ETL done, because very clean and standardize data stream (with the log after) now becomes very simple. 2.6 Log files and events log as the core architecture, as well as an additional benefit, it is easy to implement without coupling, event-driven system. The traditional way to capture user activity and system changes, such information is written to a text log, and then extracted into a data warehouse or hadoop cluster aggregation and processing of this and similar data warehouse and ETL aforementioned issues: data highly coupled with the data warehouse. In Linkedin, based kafka built event data processing system. For a variety of action defines hundreds of types of events, calls and applications exception from PV, catch the foot of the user for advertising (ad impressions), Lebron James Shoes search, to the service, and so on. In order to realize the benefits of the above event-driven system, look at a simple example of the event: the job opportunities page, provide an opportunity. This page should only be responsible for what impressions should not be too much to include other logic. However, you will find the site of a considerable size, to do this, it's easy to make more and more opportunities for independent and display logic dragged. For example, we want to integrate the following system functions: 1) We need to send data to the data warehouse hadoop and do offline processing. 2) We need to count page visits, in order to ensure that the browser is not to grab what some Web content. 3) We need to aggregate the information on this page browser, presenting opportunities for publishers in the analysis page. 4) We need to record the history of a user of this page to make sure that we provide this valuable experience good any suitable job opportunities for this user, rather than this over and over again to show the user a Opportunities (think his wife was not at home to play the game now, and that the red, green and blue flashing effects, with that the world's best dance DJ wind, or that swing the focus of the cause of the peak and Qi X small skirt of girls, and then go in only to find is the title of the party's ad now!). 5) Our recommendation systems need to record your browsing history on Air Jordan 12 this page, in order to properly track the popularity of this work opportunities. Soon, the only logical opportunity to show the page, it will become complicated. When we have this opportunity to show an increase in the mobile terminal, the forced migration of the logic in the past, which in turn exacerbated the complexity. Not finished, tangled thing is that the engineers responsible for handling this page, you need to have knowledge of other systems, in order to ensure that those functions of the 180-159220 Nike LeBron 7 VII Soldier White Black Running Shoes above can be correctly integrated. This is just a very simple example, in practice, the situation will only become more complicated. Event-driven can make it easier. Responsible for rendering the opportunities page, only showing and presenting opportunities and record some relevant factors, such as the attributes of jobs, who viewed this page, as well as other useful information related to the presentation. Page does not need to keep the knowledge and understanding of other systems, such as recommendation systems, security systems, opportunities for the publisher's analysis systems, as well as data warehouse, all of these systems need only a subscriber, subscribe to this event, and then make them independently their treatment can be, but does not require page presents opportunities because new subscribers or consumers join and make changes. 2.7 Building Scalable log isolated publisher and subscriber is not new, but to ensure multiple subscribers can process messages in real time, and simultaneously ensure scalability, the log system, is a more difficult task. If the log does not have to build fast, low overhead and scalability, then the system is built on top of this log of all the good have no chance. Many people may think that log system is a distributed system in a very slow and heavy job cost, and is only used to deal with some similar ZooKeeper more suitable for handling metadata and other information. But Linkedin now (Note: 2013), in kafka 60 billion per day in the treatment of different messages are written (if mirrored data center operator, then the tens of billions of article writing). Jay brother how they do it? 1) log split (partitioning the log) 2) to read and write by batch optimize throughput 3) Avoid unnecessary data copying by the log was cut into a plurality of partition to provide expanded capability: 1) Each partition is ordered The log, but no global order between partitions. 2) to write messages which partition is completely controlled by the writers, by dividing in accordance with a certain type of key (eg user_id). 3) division makes additional operations log, you can do in between slices (sharding) to coordinate the conduct, while ensuring the size of the system throughput and kafka cluster linear. 4) Although there is no global order (in fact, hundreds of thousands of consumers or subscribers, discuss their global order are generally not what value), but kafka provide such a guarantee: In what order will the sender of a message to a partition, this partition submit news is out and in what order (the order into what, in what order out). 5) Each partition is configured in accordance with the number of good copy, if a leader node to hang up, the other nodes will become the new leader. 6) a log, with file systems, linear read-write mode can be optimized, will write a small log can be composed of larger, high throughput operation. Kafka Air Jordan 1 did in this matter is very fierce. Batch is used in a variety of scenarios under, such as the client sends data to the server, to write data to disk, data replication between servers, the data is transmitted to the consumer, as well as other scenes sure to submit data. 7) Finally, kafka in memory log, disk log, network transmission log, using a very simple binary format, in order to facilitate the use of a variety of optimization techniques, such as zero-copy data transmission technology (zero-copy data transfer). Many optimization techniques, coming together, allowing you full, even in the case of memory, but also to read and write data in accordance with the maximum capacity of a disk or network can provide. 2.8 Logs and real-time processing you think Jay brother provided such a beautiful way to copy data to replicate opinions finished? You! Wrong! It! Log is another way of saying streams, logs the core stream processing. 2.8.1 What is stream processing Jay brother that: 1) a continuous stream processing is data processing infrastructure. Calculation Model 2) stream processing, like MapReduce or other distributed processing framework, but need to ensure low latency. 3) collect data batch type mode, resulting in a batch-type data processing mode. 4) continuous data collection mode, causing a continuous data processing mode. 5) Jay brother told a US Census way to explain the batch. In linkedin, whether live or database changes are continuous. Batch processing of data by day, and continuously calculates a window similar to the day. Therefore, stream processing is a process: 6) In the processing of data, with the concept of a time, not necessary to maintain a static snapshot of the data, so it can under the user-defined frequency, output, without having to and other data sets to reach some kind of 'end' of the state. 7) In this sense, stream processing is a batch of a generalization, and taking into account the prevalence of real-time data, which is a very important generalization. 8) Many commercial companies can not establish a stream processing engine, often because they can not establish flow data collection engine. 9) across the gulf stream processing infrastructure services and real-time response style between offline batch. 10) Log system to solve a lot of stream processing mode of the key issues, the biggest problem is how to multiple subscribers under real-time mode, providing issues of available data (stream data collection). 2.9 data stream processing flow pattern of the most interesting in that it expanded what is the data source (feeds) the concept. Whether it is raw data logs, feeds, or event data records line by line, all from the active application. However, the stream processing can also let us deal with data from other feeds, the data and raw data, consumer opinion, no different, and these derived feeds can contain any degree of complexity. A stream processing tasks, it should be this: read data from logs, writes the output logs or other systems. Input and output as logs, communication these processes themselves, and other processes, forming a map. In fact, the core of the system to log, allowing you to companies or organizations in data capture, conversion and data flow, as a series of logs and combine writing process performed thereon. A stream handler, in fact, do not have very large on: You can be a process or a set of process treatment, however, in order to facilitate the management process used code can provide some additional infrastructure and support. Introduction of logs has two purposes: 1) to ensure that the data set can support multiple subscribers mode, and orderly. 2) can be applied buffer. This is important, in asynchronous data processing process, if the producer of the upstream data faster, consumers could not keep pace, in this case, either the process of blocking the process, or the introduction of a buffer, either discard the data. Discard the data does not seem a good choice, while blocking treatment process, will make the process map all the data stream processing process stuck. The log, is a great, big, very large buffer, which allows the process to restart the process, so that after a process fails, the process does not affect the flow pattern of other processes. This is a huge organization to spread data stream is very critical, because different teams have different processing tasks, clearly not a task because an error occurs, the Air Max for Women entire stream processing processes are stuck. Storm and Samza is such a stream processing engine, and can use kafka or other similar systems as their log systems. (Note:. Jay brother quite fierce, before kafka, after the samza) 2.10 stateful real-time processing many stream processing engine is stateless, a form of a recording, but many use cases require a certain size in stream processing complex counts within the time window, aggregations and joins operations. For example, click stream, join user information. So, this use case, we need the support of the state. In the local processing of data, the need to maintain the status of a data. The question is how to maintain the correct state in case of processing may hang? The state maintained in memory may be the easiest, but give way and crash. If the maintenance status only within a certain time window, hang up or when a failure occurs, the process can be directly back to the starting point of the window retreated to play, but if the window is 1 hour so long, this may not be feasible. There is a simple way is to state the existence of a remote storage system or a database, but this will cause loss of data and produce localized network data back and forth between a lot (network round-trip). Recall that in the database mentioned in the above table and log duality. A stream processing component, you can use the local store or index to maintain state: BdbLeveldbLuceneFastbit by recording changelog on the local index, used to restore the state after crash. This mechanism, in fact, discloses a generalization can be stored as arbitrary index types, and the input stream is divided at the same time (co-partitioned) state. When dealing with process crashes, it can restore the index from the changelog, log will serve as a local state into some kind of role-based incremental backup time records. This mechanism also provides a very elegant capacity: the state of the process itself can also be recorded as a log, it is clear that other processes can subscribe to this state. Combined database log technology for data integration this scenario can often make a very powerful thing: the log extracted from the database and indexed in a variety of stream processing system, so, unlike Event streams join to become possible. 2.11 Log merger Obviously, with the log Air Jordans Men's records the total amount of full-time status change information, unlikely. Kafka uses a log merge or log garbage collection techniques: 1) For event Air Jordan 13 data, kafka leaving only a time window (which can be arranged in a few days time, or press space to configure.) 2) For keyed update, kafka using compression technology. Such log, can be used in another system via replay technology to reconstruct the original state of the system. If you keep the full amount of full-time logs, over time increase, data will become increasingly large, the reproduction process will be longer and longer. Kafka is not simply discard the old log information, instead of using the combined approach, discard obsolete records, for example, the primary key of a message recently been updated. 2.12 System 2.12.1 build a distributed system Log, distributed database systems in the data stream, and data integration in the role is the same: to maintain data consistency abstract data stream provides data recovery capabilities you can use throughout the institutions systems and data flow, as a single distributed database.