In digital advertising, it is of the utmost importance to show the highest-performing advertisements to your thousands or even millions of visitors at the right time. In order to be competitive in such an environment a great demand is placed on high performing architectures.
We will take a close look in this post at the tech stack of Billy Mobile, one of Spain’s leading ad-networks. Billy is an example of how complex and often seemingly contradictory architectural requirements can be realized through big data technology, while delivering the highest possible performance.
V for Velocity
As a digital marketplace that optimizes traffic in the mobile advertising industry, Billy Mobile’s architecture handles the traffic between thousands of advertisers and publishers. Our ad-servers account for tens of thousands to millions of visits during a single day. Together, they are producing around 60 gigabytes of data.
This entails a couple of problems: High data throughput may contradict rapid access and response time. Small chunks entering at a massive scale could unexpectedly congest pipes.
So, how can one avoid these negative effects using the least necessary hardware resources? We will address this by going through the architectural components step-by-step.
Minimizing response time
Let’s start with our first important architecture component. The Ad-Manager Service is the initial entry point of a long chain of decision making and data processing. It receives data from one of our publishers, which, in turn, receives visitor requests. Our publishers want to know which offer is the best for the incoming traffic.
To answer this question, we have to match a set of variables (targeting), provided by the incoming visitor, with the offers in our databases. These offers were defined by our advertisers. Typical targeting variables are connection type, country, device type and so on. On average, the Ad-Manager has to answer about 2,000 such visitor requests per second without losing a single one.
The billy mobile architecture
Ad-Manager is a Java application. It is a HTTP server based on the Netty framework. Netty is an asynchronous event-driven network application. It allows for rapid development of maintainable high-performance protocol servers and clients. Internally, it is orchestrating a bunch of microservices. For example, an external provider informs us about the location of a visitor by matching their IP address to the site. Such data is decisive during the which-offer-to-show decision-making process.
So, the netty-based Ad-Manager receives incoming HTTP requests from our customers’ servers. Based on the nature of advertising and the underlying business model, those requests are then redirected to a specific endpoint. In this routing layer, each endpoint usually receives the data of a different offer category or type. This is, for instance, incoming traffic of offers with the type “banners” or “links”. From there, we start processing the request, extracting all the information we can. Our objective is to predict the best advertisement that we can answer back to the publisher in less than 20 ms (percentile 95).
Balancing heavy load
To achieve a rapid response time and speed, we have set up a bunch of servers that run Ad-Managers. This increases both system availability and scalability. Being a computer cluster, though, the load balancing through all of the front servers is achieved via HA Proxy. This application is a reliable, high performance TCP/HTTP load balancer. HA Proxy has proven to achieve between 15,000 and 40,000 hits per second. It is capable of saturating a two gigabit connection with a relative low-end hardware (dual core xeons).
So, it makes sense for us to use this high-throughput technology without having to worry about it being a bottleneck. On top of HA Proxy, we distribute the traffic through the HA Proxy servers via a DNS balancer with a round robin and fail-fast strategy.
One challenge was storing cookie and user-related data with a single state on all Ad-Managers in the cluster. To manage state and share memory between the fronts, we use Redis Cluster. Redis is an in-memory key-value (data structure) store. It provides several good features that let us manage this huge amount of in-memory data, such as the auto-expire function of keys. It is simply a timeout function that invalidates keys after a certain amount of time. This allows us to reduce the amount of memory used by adjusting the Time To Live (TTL). Also, the amount of data structures (such as sorted sets and hashes), helps improve the performance of some queries without having to deal with huge network payloads and client processing.
Leveraging high data throughput to storage
A response from our Ad-Managers contains the best-performing advertisement suitable for one particular visitor to one publisher’s server. However, all of the responses that our Ad-Manager sends back to the visitors (including further contextual data) are also copied into a Apache Kafka Cluster. Kafka can be seen as a distributed log system; it is horizontally scalable, fault-tolerant, wicked fast, and runs in production in thousands of companies. Kafka is also proven to ingest millions of records per second in relative low-end hardware.
However, in our case, it is more appropriate to think about a queue system. This is where Ad-Manager and other components send all of the messages that our systems generate. Kafka keeps them safe until we can ingest them with Apache Storm for further processing (more below). Kafka allows us to temporarily decouple front- and back-end architectures. This is extremely useful when components in the back-end undergo maintenance work.
The best metaphor for imagining Kafka as a queue log mechanism is that of a water-tap. Kafka and Storm are connected by a hose. Kafka is the mechanism that prevents the water (data) in the hose from escaping. What happens to a hose (as to Kafka), though, when the tap is closed? It inflates through the pressure of the incoming water (data). However, as long as we open the tap in a timely manner, the pent-up data, like water, flushes out in massive quantities.
In reality, however, Kafka is not sending data; instead, the next player, Apache Storm, is ingesting the data from Kafka. Storm is a free and open source distributed real-time computation system. When Storm fetches data from Kafka, it usually stores it into Hive for good, with a minimum delay. Storm is fast: A benchmark clocked it at over a million tuples processed per second per node. Storm 1.0 is even faster.
The code is distributed in Storm topologies. A topology is basically a graph with two types of nodes: Spouts, a source of stream data (for example, the Kafka Spout, which reads from Kafka), and Bolts, which applies some processing into the stream of data. Thus, the topology is a top-level abstraction of the Storm stream-processed data flow. The data flows from the front via Kafka and Storm to other components. Usually, in the end, it is being inserted into Apache Hive as the main data mass storage (through the Hive Bolt, as you have probably guessed).
Queriability through structure
Before talking about Hive, we should mention Apache HBase. In general, HBase is a key-value big data store. In billy, we use HBase as an entity cache. We store the impressions and other data entering into our systems. As a result, we are able to join information between impressions and conversions to the corresponding clicks and conversions once they occur.
Remember: Chronologically speaking, impressions happen first, followed in minor quantities by clicks and other conversions. So, when to use HBase? Use HBase when you need random, realtime read/write access to your Big Data (some tens of TBs every month). Also, it couples perfectly with the hadoop ecosystem, being able to query its data through the variety of processing frameworks available (hive, pig, mapreduce, spark, etc.)
Finally, as mentioned, we store the data in Apache Hive. Apache Hive is the SQL layer on top of Hadoop. This makes the data queryable for further analysis, albeit those queries process relatively slowly. Usually, when it comes to Hadoop, everything is batch, immutable, and slow (although big data! yay!), thus making it hard to achieve a near-real-time approach for ingesting data. This is a hurdle when making data available as fast as possible to our data-science guys, commercial team, and also to our prediction algorithm. There are a variety of solutions for achieving some kind of mutating data ability. However, all of them were very hard to maintain and difficult to code.
Fortunately for us, Apache Hive has recently included an amazing feature: ACID transactions and Streaming API. It allows us to continuously add more data in the form of delta micro-batches, without losing availability or consistency. The magic happens in the background, completely managed and without hassle for our data. Hive stores every micro-batch as a delta-file. Every now and then, when we have enough of those files, the compactor comes in action and merges them into the base data. The most important part: The data is available as soon as it is committed (stored as a delta file).
One can see that every component in the architecture represents a necessary leverage in the data flow. From beginning to end, the data processing speed slows down, but increases in terms of structure and the ease of querying it. At first, massive and rapid response time is necessary, as it is necessary to secure data integrity while processing. In the mid-level, Storm and Hbase guarantee a relatively quick distribution and access to data for response processing. This data, however, is less structured, making it hard to extract for further data evaluation and analysis. This purpose is fulfilled by Hive on ACID. Ultimately, no single component could solve the inherent challenges that Billy Mobile advertising is facing. However, the right orchestration of components in the Billy Mobile architecture achieves this.