Apache Pulsar at Yahoo!JAPAN

head img
Nozomi Kurihara
September 6, 2019

Apache Pulsar at Yahoo!JAPAN

Yahoo! JAPAN is a Japanese internet company originally formed as a joint venture between Yahoo! and SoftBank. The web portal of Yahoo! JAPAN is the most popular website in Japan, and its internet services are almost dominant in the country.

The following figure illustrates the scale of Yahoo! JAPAN from 3 dimensions. The first one is the number of services. We run more than 100 services in Yahoo! JAPAN. The second one is the number of servers we used for running these 100+ services. We have more than 150,000 servers (mostly bare-metal servers) running 24x7 for serving the 100+ services. The last one is the total number of page views (PV) per month. It is more than 70 billions per month back in 2017. As you can see Yahoo! JAPAN runs internet services at large scale.

Challenges

As you can see, running Yahoo! JAPAN at large scale is a huge challenge. It requires high performant and scalable services for handling large number of users. The system has to be designed for multi-tenancy in order to serving large number of services. In some cases, we have to deal with sensitive or mission-critical messages, and thus requiring durability. Besides, geo replication is a strong requirement since Yahoo! JAPAN has multiple data centers. We need to deal with all these requirements and run our services at scale. One of our biggest challenges is to find a stable and scalable messaging platform to meet those requirements, and ensure data integrity.

Why we choose Apache Pulsar

In order to deal with these fabulous requirements, we began to investigate different messaging platforms.

Apache Pulsar vs Apache Kafka

Before adopting Apache Pulsar, we’ve compared Apache Pulsar with Apache Kafka, today’s different messaging standards.

The main difference between the two systems is data distribution. Briefly, Pulsar keeps data in a better balanced manner than Kafka. Let me explain the details as illustrated in the following figure. Let’s consider the case: the number of partitions is 1, the number of storage nodes is 3, and the replication refactor for the data is 2. In Kafka, all the data in the partition is kept by one leader broker (Broker 2). And they will be replicated to other replica broker (Broker 1). Broker 3 is neither leader or replica, so it doesn’t have any data. In Kafka, since all data in one partition is held by one broker, so the capacity of the partition is limited by the capacity of one broker. When you want to scale it, data rebalancing is needed; otherwise, data distribution could be imbalanced. On the other hand, in Pulsar, data in partition is divided into more granular units called segments. And they are evenly distributed and balanced across all bookies. Thus the capacity is not limited by a single bookie node. Moreover, data rebalancing is not needed. As a result, Pulsar is more flexible and easier to scale than Kafka.

Another difference is geo-replication. Kafka provides a tool called MirrorMaker to handle geo-replication, but you need to manage it as an extra process. On the other hand, geo-replication in Pulsar is a built-in feature, so we don’t need extra process.

OpenMessaging Benchmark

OpenMessaging Benchmark is a suite of tools that can easily compare various messaging systems. According to their report, Apache Pulsar achieves better throughput and latency than Kafka in performance. The blue one represents Pulsar and the red one represents Kafka. The left graph shows throughput, and the upper one is better. The right graph shows latency, and the lower one is better.

Why Apache Pulsar best fits

Since Apache Pulsar is reported to be better than Apache Kafka, we decide to explore it further. We have summarized all the points that why Pulsar fits best for our scale.

High performance

Apache Pulsar achieves high throughput and low latency even if a huge number of topics and high reliability is required. For example, at Oath Inc., there are 2.3 million topics, and 100 billion messages per day are processed. Message loss is not acceptable, and order needs to be guaranteed. In addition to these requirements, Apache Pulsar can achieve 1,000,000 throughput (msg/s) and 5 ms latency.

Scalability

Apache Pulsar is really a scalable system. To increase the capacity, all what we have to do is just to add servers. Moreover, since the serving layer and storage layer are separated, we can flexibly add brokers or bookies according to the data route. If we need more serving capacities, we just add brokers. If we need more storage capacity, we just need to add bookies.

Multi-tenancy

Multi-tenancy means multiple services can share one Pulsar system. Each service and application can use Pulsar as a ‘Tenant’. So different services do not need to maintain their own siloed messaging system. Apache Pulsar provides different authentication and authorization mechanisms which can be used for protecting messages from interception. We can configure and share it for each namespace or topic to protect our messages. As a result, a lot of services can run on only one Pulsar system. It reduces the maintenance and labor costs.

Geo-replication

In Yahoo! JAPAN, we run services over multiple datacenters. Pulsar provides the built-in geo-replication for replicating messages between datacenters. It handles disaster recovery, improves our service quality. More importantly, it is a built-in mechanism, and is easy to enable and use.

Pulsar Functions

Pulsar Functions is lightweight compute framework, such as AWS lambda or Google Cloud Functions. We don’t need to launch extra systems, such as Apache Heron, Apache Storm, Apache Spark, or other similar systems. All we have to do is to implement our logic and deploy it to Pulsar cluster. Java, Python, and Go are supported in Pulsar Functions.

Pulsar on Kubernetes

Many users use Kubernetes, so do we. It is easy to deploy Pulser in Kubernetes clusters. For example, the following screenshot shows that Pulsar is working on Kubernetes engine.

After careful investigation, we’ve found that Apache Pulsar not only has better performance than Apache Kafka, but also meets all of our requirements, and it is easy to deploy on Kubernetes, so we finally decide to adopt Apache Pulsar as our internal messaging platform.

How do we use Apache Pulsar

We have successfully adopted and run Pulsar on production for many years. Let me use a few use cases to explain how we use Apache Pulsar in Yahoo! JAPAN.

This is a simple illustration of the system architecture in Yahoo! JAPAN. We have two data centers: the west and the east data centers. In each data center, we have brokers, bookies, ZooKeepers and WebSocket proxy servers. And also we collect metrics by Prometheus and visualize them by Grafana.

We use Prometheus to monitor the number of topics, producers and consumers, as shown in the following graph.

Self-service tool

Meanwhile, we have developed a tool to create and manage tenants, namespaces and topics. Through UI, you can create tenants, namespaces, and configure the settings by yourself. Currently, this UI is Yahoo! JAPAN internal, not open source, and in Japanese. We can create tenant and namespace, and see topic stats such as throughput, average message size and so on.

Case 1: Notification of content update

We use Apache Pulsar as a notification service system. Various contents files (for example weather, map, or news data) are pushed from partner companies to Yahoo! JAPAN. Services want to know the update of these contents. Thus services serve files as topics. When contents are updated, notification messages are sent to the topic. Once services receive a notification, they fetch the content files from file server.

Case 2: Job queuing in mailing service

We use Apache Pulsar as an asynchronous job queuing. Indexing of mail can be a heavy job. So letʼs execute it asynchronously. At first, producer in the Mail BE server register jobs to Pulsar. Consumers take jobs from Pulsar at their own pace. And if it fails, it will re-register the job.

Case 3: Log pipeline

We use Apache Pulsar to collect logs. In Yahoo! JAPAN, almost all services and applications are running a PasS platform (like Heron) or CaaS platform (like Kubernetes). And then we want to collect logs from them. First, these logs are published to Pulsar. And once logs are published, they are divided into some topics. According to service and final destinations by using Pulsar Functions, logs are finally sent to other databases or platforms, such as HBase, Prometheus, Twilio. Thus we are running this huge log collection architecture.

Future plans

We are developing two open source projects: Pulsar node.js client and Admin WebUI.

With node.js client, node.js users can easily use Pulsar. It is based on C++ Client. We have implemented basic functionalities (producer, consumer, reader), test codes and performance scripts. We plan to publish to npm registry, fix release flow, and support more features, such as multi-topic consume, unack and so on.

Apache Pulsar Admin WebUI is under development. With this UI, administrators can easily and intuitively manage Pulsar topics. We recreate the tool by using Vue.js and Express. We have implemented basic pages (tenants, namespaces, topics and so on). We will make the repository open source soon, develop advanced commands (such as unload, skip-messages) and send authentication to Broker.

Conclusion

In conclusion, Apache Pulsar is a fast, durable, and scalable pub-sub messaging system, and has useful built-in features, such as geo-replication, multi-tenancy, Pulsar Functions, and so on.

We have adopted Apache Pulsar on production for years. Pulsar has shown great stability. We keep watching the news, updates, and activities in the Pulsar community and leverage the new features for our use cases.

The Pulsar community is still young, but there is already a fast-growing tendency of adopting Pulsar to replace many legacy messaging systems.

Reference

Do you like this blog? Share it now

twitter linkedIn