January 20, 2017 at 12:19Spotify: Migrating the Event Subsystem to Google Cloud (Part 1)
- Transfer
Whenever a user performs an action in a Spotify client - such as, for example, listening to a song or searching for an artist - a small amount of information, an event, is sent to our servers. Event delivery, the process of safe and reliable transportation of information from customers around the world to our central processing system, is an interesting task. In a series of these articles, we will consider some of the solutions that have been implemented in this area. To be more precise, we’ll look at the architecture of our new event delivery system and explain why we decided to deploy it to Google Cloud.
In this first article, we will explain how our current event delivery system works and talk about some of the lessons we learned while working with it. In the next- consider the creation of a new system and why we chose Cloud Pub / Sub as a transport mechanism for all events. In the third and final article, we will explain how we work with all events using DataFlow, and how quickly this all happens.
Events distributed through our delivery system have many uses. Most of our solutions in product design are based on the results of A / B tests, and they, in turn, must rely on large and accurate data. The Discover Weekly playlist , launched in 2015, quickly became one of Spotify’s most used features. It is created based on music playback data. Year in music , Spotify Partyand many other Spotify features are also data-based. In addition, Spotify data is one of the sources for compiling Billboard tops .
Our messaging system is one of the fundamental parts of the Spotify data infrastructure. A key requirement for it is the delivery of all data with a predictable delay and availability for our developers through a well-described interface. Usage data can be described as a set of structured events formed at some point in time as a response to some predetermined actions.
Most of the events that Spotify uses are directly generated by Spotify clients as a response to certain user actions. Whenever an event occurs in the Spotify client, information about it is sent to one of the Spotify gateways, which writes it to the system log. There he is assigned a time stamp, which is used in the message delivery system. In order to guarantee a certain delay and completeness of message delivery, it was decided for the event to use the log label (syslog timestamp), and not the client, since we do not have control over the event before it got to our servers.
In the case of Spotify, all data must be delivered to a central Hadoop cluster. The Spotify servers on which we collect data are located in several data centers on two continents. The bandwidth between our data centers is a scarce resource and it is necessary to treat data transmission with special care.
The data interface is determined by the location of the data in Hadoop and the format in which it is stored. All data delivered by our service is recorded in Avroformat in HDFS. The delivered data is divided into sections (partitions) for 60 minutes (hour). This is a relic of the past when the first message delivery system was based on the scp command and hourly copying syslog files from all servers on Hadoop. Since all data processing processes in Spotify today are based on hourly data, this interface will remain with us in the foreseeable future.
Most data processes in Spotify read data from a watch assembly only once. The output values of some processes can serve as input to others, thus forming long chains of transformations. After the process has processed the data for an hour, it no longer carries out any checks in this original hour for changes. If the data has changed, the only way to reproduce these changes further is to manually restart all the relevant tasks (and related tasks) for this particular interval (hour). This is an expensive and time-consuming process, which is why we put forward such requirements for the message delivery service and after providing the hour set we can no longer supplement any data in it. This problem, known as the data completeness problem, contrasted with the requirement of minimum delay in data processing. An interesting point of view on the problem of data completeness is presented inGoogle Dataflow Report .
Our initial messaging system was built on top of Kafka 0.7.
In it, an event delivery system is built around the abstraction of hourly files. It is designed to stream log files that contain events from service machines to HDFS. After all the log files are transferred to HDFS for a certain hour, they are converted from text with tabs to the Avro format.
When the system was first created, one of the missing features of Kafka 0.7 was the ability of the Kafka Broker cluster to work with reliable persistent storage. This influenced the adoption of an important design decision - not to maintain a constant state between the data producer, Kafka Syslog Producer and Hadoop. An event is considered securely saved only when it is written to a file on HDFS.
The problem with the reliable existence of the event only inside Hadoop is that the Hadoop cluster becomes a single point of failure for the message delivery system. If Hadoop fails, the entire delivery system will stop. To cope with this, we need to make sure that we have enough disk space on all the services from which we collect events. When Hadoop returns to service, we need to “catch up” with its status, transferring all the data as quickly as possible. Recovery time is mainly limited by the bandwidth that we can use between our data centers.
A Producer is a daemon that runs on every host from which we want to send events to Hadoop. It tracks log files and sends log packages to Kafka Syslog Consumer. The producer does not know anything about the type of event or the properties that it may have. From his point of view, an event is just a set of lines in a file and all lines are redirected to the same channel. This means that events of all types contained in one log file are also transmitted through one channel. In such a system, Kafka topics are used as channels for transmitting events. After the Producer sends the logs to the Consumer, he needs to wait for confirmation (ACK) that Consumer successfully saved the log lines in HDFS. Only after the producer receives ACK for the sent logs, he believes
For events, to get from Producer to Consumer, you need to go through Kafka Brokers and then Kafka Groupers. Kafka Brokers is a standard component of Kafka, and Kafka Groupers is a component that we wrote. Groupers processes all event streams from local data centers and then publishes them again compressed, effectively grouped in one topic, which is then pulled by Consumer.
The Extract, Transform and Load (ETL) task is used to convert data from a simple format, separated by tabs, to the Avro format. This process is a normal Hadoop MapReduce job, implemented using the Crunch framework , which works with hourly sets. Before starting work with a certain hour, he needs to make sure that all files are completely transferred.
All Producers are constantly sending check marks that may contain end-of-file tokens. These tokens are sent only once, when Producer concluded that the entire file was securely stored on Hadoop. The state (or “survivability”) monitor constantly polls our service discovery systems in all data centers about which service machines were running at a particular hour. To check whether all files were finally transferred in this hour, ETL compares information about the servers from which it should expect data with end-of-file markers. If the ETL determines discrepancy and incomplete data transfer, then it delays data processing for a specific hour.
In order to be able to maximize the use of existing mappers and reducers, ETL, which is a common task of Hadoop MapReduce, you need to know how to shard input data. Mapers and reducers are calculated based on the size of the input data. Optimal sharding is calculated based on the number of events continuously coming from Consumers.
One of the main problems associated with this design is that local Producers must ensure that data is stored in HDFS in a central location before it can be considered reliably delivered. This means that the Producer server on the west coast of the United States needs to know when data is being written to disk in London. Most of the time it works just fine, but if the data transfer slows down, it will cause delays in delivery, which will then be difficult to get rid of.
Compare this with options when the service point is located in a local data center. This simplifies the design of the Producer, since, usually, the network between the hosts in the data center is very reliable.
Abstracting from the problems, we were quite pleased with the system, which can reliably deliver more than 700,000 events per second from around the world. The redesign of the system also gave us the opportunity to improve the software development process.
By sending all events together through one channel, we lost the flexibility of managing event flows with different quality of service (QoS). This also limited the work in real time, since any process working in real time had to transfer its data through a single channel in which the entire stream goes, and filter out only the necessary one from it.
The transfer of unstructured data adds unnecessary delay, as it requires additional ETL conversion. Currently, ETL work adds about 30 minutes of delay to event delivery. If the data were sent in the Avro format, then they were immediately available when recording on HDFS.
The need for the sender to track the end of the hour also caused problems. For example, if a machine dies, it cannot send a message about the end of the file. If the end-of-file marker is lost, then we will wait forever until this process is interrupted manually. As the number of cars grows, this problem becomes more and more urgent.
The number of delivered messages in Spotify is constantly increasing. As a result of increased loads, we began to experience more and more problems. Over time, the number of outages began to worry us. We realized that neither we, nor the system, can no longer cope with the increased load. Just in the next article we will tell you about how you decided to change our system.
The number of messages processed by our system at a particular point in time.
In this first article, we will explain how our current event delivery system works and talk about some of the lessons we learned while working with it. In the next- consider the creation of a new system and why we chose Cloud Pub / Sub as a transport mechanism for all events. In the third and final article, we will explain how we work with all events using DataFlow, and how quickly this all happens.
Events distributed through our delivery system have many uses. Most of our solutions in product design are based on the results of A / B tests, and they, in turn, must rely on large and accurate data. The Discover Weekly playlist , launched in 2015, quickly became one of Spotify’s most used features. It is created based on music playback data. Year in music , Spotify Partyand many other Spotify features are also data-based. In addition, Spotify data is one of the sources for compiling Billboard tops .
Our messaging system is one of the fundamental parts of the Spotify data infrastructure. A key requirement for it is the delivery of all data with a predictable delay and availability for our developers through a well-described interface. Usage data can be described as a set of structured events formed at some point in time as a response to some predetermined actions.
Most of the events that Spotify uses are directly generated by Spotify clients as a response to certain user actions. Whenever an event occurs in the Spotify client, information about it is sent to one of the Spotify gateways, which writes it to the system log. There he is assigned a time stamp, which is used in the message delivery system. In order to guarantee a certain delay and completeness of message delivery, it was decided for the event to use the log label (syslog timestamp), and not the client, since we do not have control over the event before it got to our servers.
In the case of Spotify, all data must be delivered to a central Hadoop cluster. The Spotify servers on which we collect data are located in several data centers on two continents. The bandwidth between our data centers is a scarce resource and it is necessary to treat data transmission with special care.
The data interface is determined by the location of the data in Hadoop and the format in which it is stored. All data delivered by our service is recorded in Avroformat in HDFS. The delivered data is divided into sections (partitions) for 60 minutes (hour). This is a relic of the past when the first message delivery system was based on the scp command and hourly copying syslog files from all servers on Hadoop. Since all data processing processes in Spotify today are based on hourly data, this interface will remain with us in the foreseeable future.
Most data processes in Spotify read data from a watch assembly only once. The output values of some processes can serve as input to others, thus forming long chains of transformations. After the process has processed the data for an hour, it no longer carries out any checks in this original hour for changes. If the data has changed, the only way to reproduce these changes further is to manually restart all the relevant tasks (and related tasks) for this particular interval (hour). This is an expensive and time-consuming process, which is why we put forward such requirements for the message delivery service and after providing the hour set we can no longer supplement any data in it. This problem, known as the data completeness problem, contrasted with the requirement of minimum delay in data processing. An interesting point of view on the problem of data completeness is presented inGoogle Dataflow Report .
Original message delivery system
System architecture
Our initial messaging system was built on top of Kafka 0.7.
In it, an event delivery system is built around the abstraction of hourly files. It is designed to stream log files that contain events from service machines to HDFS. After all the log files are transferred to HDFS for a certain hour, they are converted from text with tabs to the Avro format.
When the system was first created, one of the missing features of Kafka 0.7 was the ability of the Kafka Broker cluster to work with reliable persistent storage. This influenced the adoption of an important design decision - not to maintain a constant state between the data producer, Kafka Syslog Producer and Hadoop. An event is considered securely saved only when it is written to a file on HDFS.
The problem with the reliable existence of the event only inside Hadoop is that the Hadoop cluster becomes a single point of failure for the message delivery system. If Hadoop fails, the entire delivery system will stop. To cope with this, we need to make sure that we have enough disk space on all the services from which we collect events. When Hadoop returns to service, we need to “catch up” with its status, transferring all the data as quickly as possible. Recovery time is mainly limited by the bandwidth that we can use between our data centers.
A Producer is a daemon that runs on every host from which we want to send events to Hadoop. It tracks log files and sends log packages to Kafka Syslog Consumer. The producer does not know anything about the type of event or the properties that it may have. From his point of view, an event is just a set of lines in a file and all lines are redirected to the same channel. This means that events of all types contained in one log file are also transmitted through one channel. In such a system, Kafka topics are used as channels for transmitting events. After the Producer sends the logs to the Consumer, he needs to wait for confirmation (ACK) that Consumer successfully saved the log lines in HDFS. Only after the producer receives ACK for the sent logs, he believes
For events, to get from Producer to Consumer, you need to go through Kafka Brokers and then Kafka Groupers. Kafka Brokers is a standard component of Kafka, and Kafka Groupers is a component that we wrote. Groupers processes all event streams from local data centers and then publishes them again compressed, effectively grouped in one topic, which is then pulled by Consumer.
The Extract, Transform and Load (ETL) task is used to convert data from a simple format, separated by tabs, to the Avro format. This process is a normal Hadoop MapReduce job, implemented using the Crunch framework , which works with hourly sets. Before starting work with a certain hour, he needs to make sure that all files are completely transferred.
All Producers are constantly sending check marks that may contain end-of-file tokens. These tokens are sent only once, when Producer concluded that the entire file was securely stored on Hadoop. The state (or “survivability”) monitor constantly polls our service discovery systems in all data centers about which service machines were running at a particular hour. To check whether all files were finally transferred in this hour, ETL compares information about the servers from which it should expect data with end-of-file markers. If the ETL determines discrepancy and incomplete data transfer, then it delays data processing for a specific hour.
In order to be able to maximize the use of existing mappers and reducers, ETL, which is a common task of Hadoop MapReduce, you need to know how to shard input data. Mapers and reducers are calculated based on the size of the input data. Optimal sharding is calculated based on the number of events continuously coming from Consumers.
The lessons
One of the main problems associated with this design is that local Producers must ensure that data is stored in HDFS in a central location before it can be considered reliably delivered. This means that the Producer server on the west coast of the United States needs to know when data is being written to disk in London. Most of the time it works just fine, but if the data transfer slows down, it will cause delays in delivery, which will then be difficult to get rid of.
Compare this with options when the service point is located in a local data center. This simplifies the design of the Producer, since, usually, the network between the hosts in the data center is very reliable.
Abstracting from the problems, we were quite pleased with the system, which can reliably deliver more than 700,000 events per second from around the world. The redesign of the system also gave us the opportunity to improve the software development process.
By sending all events together through one channel, we lost the flexibility of managing event flows with different quality of service (QoS). This also limited the work in real time, since any process working in real time had to transfer its data through a single channel in which the entire stream goes, and filter out only the necessary one from it.
The transfer of unstructured data adds unnecessary delay, as it requires additional ETL conversion. Currently, ETL work adds about 30 minutes of delay to event delivery. If the data were sent in the Avro format, then they were immediately available when recording on HDFS.
The need for the sender to track the end of the hour also caused problems. For example, if a machine dies, it cannot send a message about the end of the file. If the end-of-file marker is lost, then we will wait forever until this process is interrupted manually. As the number of cars grows, this problem becomes more and more urgent.
Next steps
The number of delivered messages in Spotify is constantly increasing. As a result of increased loads, we began to experience more and more problems. Over time, the number of outages began to worry us. We realized that neither we, nor the system, can no longer cope with the increased load. Just in the next article we will tell you about how you decided to change our system.
The number of messages processed by our system at a particular point in time.
