BigBird: Big Data Storage and Analytics at Scale in Hybrid Cloud

Partly Cloudy project is a pivotal step in replicating the on-premise HDFS data to Google Cloud Storage (GCS). While we had the HDFS file system path for the data, we did not have the corresponding GCS buckets. To achieve that resource parity in the cloud, we developed an in-house suite of services called Demigod. The Demigod suite of services are responsible for a) managing the life cycle of the GCS buckets corresponding to the on-premise HDFS file path b) managing the life cycle of the “shadow” service accounts in GCP that correspond to the on-premise UNIX users and c) assigning the appropriate permissions for “shadow” accounts and users to these buckets so that the services can read and write data to the buckets. Section 2.1.2 discusses the concept of “shadow” accounts. The working of Demigod services can be a different topic to discuss and therefore, not in scope for this paper. Since all buckets in GCS live in a global namespace, to avoid bucket names clash with other GCP users, twitter.domain is a verified bucket domain name used for Twitter’s big data storage in GCS. Therefore, for uniqueness, GCS buckets in Twitter’s GCP would end with a “twitter.domain” string. Now that we had GCS resources in GCP, our “replicator” services needed to be aware of the path conversion from HDFS to GCS. Therefore, we implemented a path conversion mechanism to logically map an on-premise HDFS path to a globally unique GCS bucket path. This conversion happens in two stages. In the first stage, the on-premise HDFS path is converted into a logical GCS path, for example, /dc1/cluster1/user/helen/some/path/part-001.lzo is converted into logical GCS path gcs/user/helen/some/path/part-001.lzo. Then in the second stage, the logical path is converted into an actual GCS bucket path viz. gs://user.helen.dp.twitter.domain/some/path/part-001.lzo. Similarly, for logs, the GCS path may be gs://log.activity-logs.dp.twitter.domain/some/path/part-001.lzo. We provision these GCS buckets under a dedicated GCP project named like “twitter-gcs-project”. This path conversion methodology helps the replication services seamlessly find a GCS bucket path for a corresponding HDFS file system path. These paths form the basis of data replication and ingestion from on-premise HDFS file system into GCS, and eventually from GCS into BigQuery.

When we started working on enabling big data storage and analytics at Twitter, we wanted to keep the user experience similar to how they interact with data stored in GCS. This would mean that we would have a single GCP project that stores all BigQuery datasets for the users and services, an approach that was similar to the data storage framework using GCS as mentioned in the above sections. Although this would have made it easier for services and users to access BigQuery data, we quickly realized that the scale of our data and operations would not fit quotas and limitations GCP has put in place for BigQuery bib11 . Every service in GCP has associated quotas and limitations in place for scalability purposes.

In our framework design, GCP Projects are a fundamental way to organize and access data. Our design recommends separating storage and compute for security adherence, resource cost chargeback, and manageability. The design principles in separating storage and compute along with human and machine accounts were largely influenced by Twitter’s compliance needs. Using this recommended project architecture for storing data provides an easy way for data registration, annotation, retention, and wipeout that would otherwise need to be solved separately by individual users and teams.

There are a few ways the services and users can ingest their data from HDFS or GCS into BigQuery. Our framework provides flexibility in ingesting the data into BigQuery including the scale of ingestion and ease of operations. We explain a few of major ways of ingestion as follows and compare them in Table 1. For the purpose of ingesting the data, our framework designates a dedicated “data load” project named “twitter-gcs-to-bq-project” and data is loaded from the GCS bucket into corresponding BigQuery dataset via this project.

Since we separated the storage and compute projects, our framework makes it seamless for the services to perform big data analytics on top of BigQuery data. The teams at Twitter heavily use machine learning to analyze the data and perform prediction analysis, BigQuery is a perfect fit for those use cases. Native support for machine learning in BigQuery enables at-scale big data analytics using BigQuery ML bib14 . BigQuery ML (BQML) enables users to create and execute machine learning models in BigQuery using SQL queries. The services and users can train, evaluate, and even serve models natively from their compute projects while reading the data stored in BigQuery storage projects. Models can be created in BigQuery using the CREATE MODEL statement. This single statement handles the gathering of the training data that is done via a SELECT statement, pre-processing the data, creating the model, training the model, and optionally evaluating the model against validation data. Once the model has been trained, its evaluation and training statistics can be viewed directly in the BigQuery UI, such as the receiver operating characteristic (ROC) curve and the confusion matrix.

CREATE MODEL ‘mydataset.mymodel’
OPTIONS (MODEL_TYPE=‘LINEAR_REG’)
AS
SELECT * FROM twitter-my-service-bq-project

Since the big data analytics operations performed over the BigQuery data are compute operations, they need CPU time to perform these operations. CPU time varies based on the type of query, the scale of the query, and data consumed by the machine learning modeling that is run on top of the data stored in BigQuery. A Slot is a unit of computational capacity that may include a combination of RAM and CPU bib17 . Multiple slots can be classified into the “buckets” called Reservations. Our framework also makes it possible for the users to consume slots based on their usage and exclusivity of the service. We create two types of slot reservations viz., default and dedicated slot reservations. To start with all the projects that run big data analytics jobs or query data from BigQuery storage projects are added to the default reservation. For example, if we have 100000 slots available in a default reservation, all projects must share this compute capacity. This begets problems like ”Noisy Neighbors” where a large analytics job may consume a major chunk of the default reservation keeping other jobs starved. To alleviate this problem, our framework also implements a dedicated reservation where an analytics job is guaranteed a dedicated compute capacity. This dedicated reservation is formed from available default reservation. Thus, if the default reservation has 100000 slots available and if an analytics job “tweet_analyzer” requires 30000 slots then for the dedicated reservation is created for the compute project that runs this analytics job. Default reservation, therefore, is left with 70000 slots that are shared among all the compute projects. Although the framework also supports purchasing extra slots real-time, Twitter currently relies on capacity forecasting to buy slots in advance.

To forecast the capacity and usage of the compute and storage quota, we monitor all the jobs, including queries and analytics, that are running at the moment. Since there are hundreds of storage and compute projects, it is not practical to run an ”agent” virtual machine in every project tracking the usage and operational statistics. Because of the large scale of the framework, we decided to use a resource at an organization node that will give us bird’s eye visibility into all compute and storage projects. We chose to use INFORMATION_SCHEMA as our basis for fetching all the metrics related to BigQuery usage. INFORMATION_SCHEMA is a series of views that provide information about BigQuery metrics like the number of datasets, jobs, jobs’ timeline, reservation assigned to a job, slots consumed by jobs, and access control. We implemented a central metrics collector agent process that queries INFORMATION_SCHEMA after a fixed interval, aggregates those metrics based on a query or analytics job, and creates the monitoring dashboards for the data admins to track. Alerts are generated on top of these dashboards and if a metric crosses a certain threshold then the data admins are alerted via a pager software. Fig. 6 showcases a small subset of monitoring dashboards for BigBird. Our framework tracks similar metrics for slots allocation, slot usage, jobs waiting for slot allocation and so on.

Similar to how compute and storage projects are monitored, we wanted a similar way to generate audit logs for every operation that is performed on the data stored in the storage projects. This includes dataset creation and deletion, data read by a reader group, data written by a shadow service account, and any access control change by an admin account. To achieve this at scale, we set up a logging sink at the organization node with the query that tracks above activities across all the compute and storage projects. The logs are stored inside a BigQuery dataset and persisted as per the compliance requirements.

Since data loading in our framework happens in two different phases viz., HDFS to GCS, and GCS to BigQuery, it is essential to have visibility into the delays for data loading. Data Watchdog is a service that analyzes the latencies in jobs that load data from one storage system into another. The monitoring of data pipeline also aids in debugging by providing visibility into what part of the pipeline the data movement is stuck and provide latency visibility across datasets by keeping track of the status of the datasets if the datasets exist in HDFS and GCS, and last time the dataset files were changed. The actual architectural discussion of Data Watchdog is out of scope for this paper. However, it plays an essential part in the monitoring of the overall framework.