MapReduce-based warehousing solutions (e.g. Hive) for big data analytics with the capabilities of storing and analyzing high volume of both structured and unstructured data in a scalable le system have emerged recently. Their e cient data loading features enable a so-called near real-time warehousing solution in contrast to those o ered by conventional data warehouses with complex, long-running ETL processes. However, there are still many opportunities for performance improvements in MapReduce systems. The performance of analyzing structured data in them cannot cope with the one in traditional data warehouses. For example, join operations are generally regarded as a bottleneck of performing generic complex analytics over structured data with MapReduce jobs. In this paper, we present one approach for improving performance in MapReduce-based warehouses by pre-joining frequently used dimension columns with fact table redundantly during data transfer and adapting queries to this joinfriendly schema automatically at runtime using a rewrite component. This approach is driven by the statistics information derived from previous executed workloads in terms of join operations. The results show that the execution performance is improved by getting rid of join operations in a set of future workloads whose join exactly ts the pre-joined fact table schema while the performance still remains the same for other workloads.
By packaging complex custom imperative programs (text mining, machine learning, etc.) into simple map and reduce functions and executing them in parallel on les in a large nished his work during his master study at university of kaiserslautern scalable le system, MapReduce/Hadoop1 systems enable analytics on large amounts of unstructured data or structured data in acceptable response time.
With the continuous growth of data, scalable data stores based on Hadoop/HDFS2 have achieved more and more attention for big data analytics. In addition, by means of simply pulling data into the le system of MapReduce-based systems, unstructured data without schema information is directly analyzed with parallelizable custom programs, whereas data can only be queried in traditional data warehouses after it has been loaded by ETL tools (cleansing, normalization, etc.), which normally takes a long period of time.
Consequently, many web or business companies add
MapReduce systems to their analytical architecture. For example,
Fatma Ozcan et al. [
Another example is the data warehousing infrastructure at Facebook which involves a web-based tier, a federated MySQL tier and a Hadoop-based analytical cluster - Hive.
Such orchestration of various analytical platforms forms a heterogeneous environment where each platform has a di erent interface, data model, computational capability, storage system, etc.
Pursuing a global optimization in such a heterogeneous environment is always challenging, since it is generally hard to estimate the computational capability or operational cost concisely on each autonomous platform. The internal query engine and storage system do not tend to be exposed to outside and are not designed for data integration.
In our case, relational databases and Hadoop will be integrated together to deliver an analytical cluster. Simply transferring data from relational databases to Hadoop without considering the computational capabilities in Hadoop can lead to lower performance.
As an example, performing complex analytical workloads over multiple small/large tables (loaded from relational data1one open-source implementation of MapReduce framework from Apache community, see http://hadoop.apache.org 2Hadoop Distributed File System - is used to store the data in Hadoop for analysis 1. Adaptively pre-joining tables during data transfer for better performance in Hadoop/Hive. 2.2
Hive
bases) in Hadoop leads to a number of join operations which
slows down the whole processing. The reason is that the
join performance is normally weak in MapReduce systems
as compared to relational databases [
To achieve better global performance in such an analytical synthesis with multiple platforms from a global perspective of view, several strategies can be applied.
One would be simply improving the join implementation
on single MapReduce platform. There have been several
existing works trying to improve join performance in
MapReduce systems [
Another one would be using heuristics for global performance optimization. In this paper, we will take a look at the second one. In order to validate our general idea of improving global performance on multiple platforms, we deliver our adaptive approach in terms of join performance. We take the data ow architecture at Facebook as a starting point and the contributions are summarized as follows: 2. Rewriting incoming queries according to changing table schema.
The remainder of this paper is structured as follows: Section 2 describes the background of this paper. Section 3 gives a nave approach of fully pre-joining related tables. Based on the performance observation of this nave approach, more considerations have been taken into account and an adaptive pre-join approach is proposed in Section 4, followed by the implementation and experimental evaluation shown in Section 5. Section 6 shows some related works. Section 7 concludes with a summary and future work.
In this section, we will introduce our starting point, i.e. the analytical data ow architecture at Facebook and its MapReduce-based analytical platform - Hive. In addition, the performance issue in terms of join is also stated subsequently. 2.1
Instead of using a traditional data warehouse, Facebook
uses Hive - a MapReduce-based analytical platform - to
perform analytics on information describing advertisement.
The MapReduce/Hadoop system o ers high scalability which
enables Facebook to perform data analytics over 15PB of
data and load 60TB of new data every day [
As depicted in Figure 1, data is extracted from two types of data sources: a federated MySQL tier and a web-based tier. The former o ers the category, the name and corresponding information of the advertisements as dimension data while the actions such as viewing an advertisement, clicking on it, fanning a Facebook page are extracted as fact data from the latter.
There are two types of analytical cluster: production Hive cluster and ad hoc Hive cluster. Periodic queries are performed on the production Hive cluster while the ad hoc queries are executed on the ad hoc Hive cluster. the advertiser information etc. The data sets originating in the latter mostly correspond to actions such as viewing an advertisement, clicking on it, fanning a Facebook page etc. In traditional data warehousing terminology, more often than not the data in the
Hive [
Hive supports a higher-level SQL-like language called HiveQL for users who are familiar with SQL for accessing les in Hadoop/HDFS, which highly increases the productivity of using MapReduce systems. When a HiveQL query comes in, it will be automatically translated into corresponding MapReduce jobs with the same analytical semantics. For this purpose, Hive has its own meta-data store which maps the HDFS les to the relational data model. Files are logically interpreted as relational tables during HiveQL query execution.
Furthermore, in contrast to high data loading cost (using ETL jobs) in traditional data warehouses, Hive bene ts from its e cient loading process which pulls raw les directly into Hadoop/HDFS and further publishes them as tables. This feature makes Hive much more suitable for dealing with large volumes of data (i.e. big data). 2.3
There has been an ongoing debate comparing parallel
database systems and MapReduce/Hadoop. In [
Moreover, as MapReduce was not originally designed to
combine information from two or more data sources, join
implementations are always cumbersome [
As Hive is built on top of MapReduce/Hadoop, the join
operation is essentially done by corresponding MapReduce
jobs. Thus, Hive su ers from these issues even though there
have been e orts [
Hadoop clusters processes dump compressing the into the Hive-Ha failures and also much load on th running the scrap avoiding extra lo any notions of str order to avoid lo database server cannot be read e data from that pa servers, there are the scrapes and b data a daily du Hadoop clusters. tables.
where the data stream processes Hadoop cluster strict delivery de Hive-Hadoop clu well as any ad h data sets. The ad run production jo 350 300 ) sce250 ( iem200 t n reu150 ga rve100 a 50 0 the periodic queries on production Hive-Hadoop cluster, a frequent column set could be extracted.
One example is illustrated in Figure 3. The frequent set of additional columns has been extracted. The column r in dimension table is frequently joined with fact table in company in the previous workloads as a lter or aggregate column, as the same for the column x in dimension table . During next load phase, the fact table is expanded by redundantly pre-joining these two additional columns r and x with it.
Depending on the statistics information of previous queries, di erent frequent sets of additional columns could be found in diverse time intervals. Thus, the fact table is pre-joined in an adaptive manner.
Assume that the additional columns identi ed in previous queries will also frequently occur in the future ones (as in the Facebook example), the bene ts of adaptive pre-join approach are two-fold:
First, when all the columns (including dimension columns) in a certain incoming query which requires a join operation have been contained in the pre-joined fact table, this query could be directly performed on the pre-joined fact table without join.
Second, the adaptive pre-join approach leads to a smaller table size in contrast to the full pre-join approach, as only subsets of the dimension tables are pre-joined. Thus, the resulting storage overhead is reduced, which plays a significant role especially in big data scenarios (i.e. terabytes, petabytes of data).
To automatically accomplish the adaptive pre-join approach, three sub-steps are developed: frequent column set extraction, pre-join and query rewrite.
In the rst phase, the statistics collected for extracting frequent set of additional columns is formated as a list of entries each which has the following form: Set : fFact; Dim X.Col i; Dim X.Col j ::: Dim Y.Col kg
The join set always starts with the involved fact table while the joint dimension columns are identi ed and capno pre-join ful pre-join 5GB
10GB data set size (a) Average Runtimes 5GB
10GB data set size (b) Accessed Data Volume performance can be signi cantly increased if workloads with the same join pattern later frequently occur, especially for periodic queries over production Hive-Hadoop cluster in the Facebook example.
However, the result of performing the same query on the data set with SF 10 size is disappointing as there is no performance gain while paying 12.5GB storage for redundancy (shown in Figure 2(b)), which is not what we expected. The reason could be that the overhead of scanning such redundant fully pre-joined tables and the high I/O cost as well o set the performance gain as the accessed data volume grows.
Taking the lessons learned from the full pre-join approach above, we propose an adaptive pre-join approach in this paper.
Instead of pre-joining full dimension tables with the fact table, we try to identify the dimension columns which occurred frequently in the select, where, etc. clauses of previous executed queries for ltering, aggregation and so on. We refer to these columns as additional columns as compared to the join columns in the join predicates. By collecting a list of additional column sets from previous queries, for example, tured from the select, where, etc. clauses or from the subqueries.
The frequent set of additional columns could be extracted
using a set of frequent itemset mining approaches [
As the table schema is changed in our case (i.e. newly generated fact table schema), initial queries need to be rewritten for successful execution. Since the fact table is pre-joined with a set of dedicated redundant dimension columns, the tables which are involved in the from clause of the original query can be replaced with this new fact table once all the columns have been covered in it.
By storing the mapping from newly generated fact table schema to the old schema in the catalog, the query rewrite process can be easily applied. Note that the common issue of handling complex sub-queries for Hive can thereby be facilitated if the columns in the sub-query have been prejoined with the fact table.
We use Sqoop3 as the basis to implement our approach. The TPC-H benchmark data set with SF 10 is adaptively pre-joined according to the workload statistics and transferred from MySQL to Hive. First, the extracted join pattern information is sent to Sqoop as additional transformation logic embedded in the data transfer jobs for generating the adaptive pre-joined table schema on the original data sources. Furthermore, the generated schema is stored in Hive to enable automatic query rewrite at runtime.
We tested the adaptive pre-join approach on a six-node cluster (Xeon Quadcore CPU at 2.53GHz, 4GB RAM, 1TB SATA-II disk, Gigabit Ethernet) running Hadoop and Hive.
After running the same TPC-H Query 3 over the adaptive pre-joined table schema, the result in the Figure 4(a) shows that the average runtime is signi cantly reduced. The join )25 BG (s lrkadoo20 iftrcxgeun15 w e10 eo m luo 5 tvaad 0 nopre-join ful pre-join adaptivepre-join nopre-join ful pre-join adaptivepre-join 350 300 )c250 (se iem200 t n reu150 rga vae100 50 0
10GB datasetsize (a) Average Runtimes 10GB datasetsize (b) Accessed Data Volume task has been eliminated for this query and the additional overheads (record parsing, I/O cost) have been relieved due to the smaller size of redundancy as shown in Figure 4(b).
An adaptively pre-joined fact table is essentially a
materialized view in Hive. Creating materialized views in data
warehouses is nothing new but a technique used for query
optimization. Since 1990s, a substantial e ort [
Cheetah [
We propose a schema adaption approach for global optimization in an analytical synthesis of relational databases and a MapReduce-based warehouse - Hive. As MapReduce systems have weak join performance, frequently used columns of dimension tables are pre-joined with the fact table according to useful workload statistics in an adaptive manner before being transfered to Hive. Besides, a rewrite component enables the execution of incoming workloads with join operations over such pre-joined tables transparently. In this way, better performance can be achieved in Hive. Note that this approach is not restricted to any speci c platform like Hive. Any MapReduce-based warehouse can bene t from it, as generic complex join operations occur in almost every analytical platform.
However, the experimental results also show that the performance improvement is not stable while the data volume grows continuously. For example, when the query is executed on one larger pre-joined table, the performance gain from eliminating joins is o set by the impact caused by the record parsing overhead and high I/O cost during the scan, which results in worse performance. This concludes that the total performance of complex data analytics is e ected by multiple metrics rather than a unique consideration, e.g. join.
With the continuous growth of data, diverse frameworks and platforms (e.g. Hive, Pig) are built for large-scale data analytics and business intelligent applications. Data transfer between di erent platforms generally takes place in the absence of key information such as operational cost model, resource consumption, computational capability etc. within platforms which are autonomous and inherently not designed for data integration. Therefore, we are looking at a generic description of the operational semantics with their computational capabilities on di erent platforms and a cost model for performance optimization from a global perspective of view. The granularity we are observing is a single operator in the execution engines. Thus, a global operator model with generic cost model is expected for performance improvement in several use cases, e.g. federated systems.
Moreover, as an adaptively pre-joined fact table is
regarded as a materialized view in a MapReduce-based
warehouse, another open problem left is how to handle the view
maintanence issue. The work from [