Thuong-Cang Phan
email: thuongcang.phan@isima.fr

Laurent D'orazio
email: laurent.dorazio@isima.fr

Philippe Rigaux
email: philippe.rigaux@cnam.fr

A Theoretical and Experimental Comparison of Filter-based Equijoins in MapReduce

Keywords: Big data, Cloud computing, Big data analysis, MapReduce, Equijoin, Bloom filter, Intersection Bloom filter

MapReduce has become an increasingly popular framework for large-scale data processing. However, complex operations such as joins are quite expensive and require sophisticated techniques. In this paper, we review state-of-the-art strategies for joining several relations in a MapReduce environment and study their extension with filter-based approaches. The general objective of filters is to eliminate non-matching data as early as possible in order to reduce the I/O, communication and CPU costs. We examine the impact of systematically adding filters as early as possible in MapReduce join algorithms, both analytically with cost models and practically with evaluations. The study covers binary joins, multi-way joins and recursive joins, and addresses the case of large inputs that gives rise to the most intricate challenges.

Introduction

Since the advent of applications that propose Web-based services to a worldwide population of connected people, the information technology community has been confronted to unprecedented amount of data, either resulting from an attempt to organize an access to the Web information space (search engines), or directly generated by this massive amount of users (e.g., social networks). Companies like Google and Facebook, representative of those two distinct trends, have developed for their own needs large-scale data processing platforms. These platforms combine an infrastructure based on millions of servers, data repositories where the least collection size is measured in Petabytes, and finally data processing software products that massively exploit distributed computing and batch processing to scale at the required level of magnitude. Although the Web is a primary source of information production, Big Data issues can now be generalized to other areas that continuously collect data and attempt to make sense of it. Sensors incorporated in electronic devices, satellite images, web server logs, bioinformatics, are considered as gold mines of information that just wait for the processing power to be available, reliable, and apt at evaluating complex analytic algorithms.

The MapReduce programming model [START_REF] Dean | MapReduce: Simplified Data Processing on Large Clusters[END_REF] has become a standard for processing and analyzing large datasets in a massively parallel manner. Its success comes from both its simplicity and nice properties in terms of fault tolerance, a necessary feature when hundreds or even thousands of commodity machines are involved in a job that may extend over days or weeks. However, the MapReduce programming model suffers from severe limitations when it comes to implement algorithms that require data access patterns beyond simple scan/grouping operation. In particular, it is a priori not suited for operations with multiple inputs.

One of the most representative such operations are joins. A join combines related tuples from datasets on different column schemes and thus raises at a generic level the problem of combining several data sources with a programming framework initially designed for scanning, processing and grouping a single input. Join is a basic building block used in many sophisticated data mining algorithms, and its optimization is essential to ensure efficient data processing at scale.

In the present paper we provide a systematic study of joins with filters for early removal of non-participating tuples from the input datasets. As known for a long time in the classical context of relational databases, early elimination of useless data is a quite effective technique to reduce the IO, CPU and communication costs of data processing algorithms. The approach can be transposed in distributed systems in general, and to MapReduce frameworks in particular.

We focus on equijoins, and examine state-of-the-art algorithms for two-way joins, multi-way joins and recursive joins. We compare, analytically and experimentally, the benefit that can be expected by introducing filters as early as possible in the data processing workflow. Our result put the research contributions in this field in a coherent setting and clarifies the stakes of combining several inputs with MapReduce.

The rest of the paper is organized as follows. Section 2 summarizes the background of the basic join operation, recalls the essentials of the MapReduce framework and intersection filters, and positions our paper with respect to related work. Section 3 presents filter-based equijoins in MapReduce. We examine twoway joins, multi-way joins, and recursive joins. Section 4 analyzes the algorithms and introduces cost models. The evaluation environment and the results are reported in Section 5. Finally, Section 6 concludes and discusses future work.

Table 1 provides a quick reference to the algorithms abbreviations used throughout the text.

2 Background and related work

Join operation

A join combines tuples from one or several relations according to some join condition 3 . A tuple that participates to the result (and therefore satisfies the join condition) is called a matching tuple in the following. Non-matching tuples Three-way join proposed by Afrati and Ullman [START_REF] Afrati | Optimizing joins in a map-reduce environment[END_REF]

cj-ifbj

Chain join using an intersection filter-based join cascade cj-bj Chain join using a Bloom join (bj) cascade cj-rsj

Chain join using a reduce-side join (rsj) cascade ocj-2wj

Optimized chain join using a two-way join cascade ocj-3wj

Optimized chain join using a three-way join (3wj) cascade rej-shaw

Recursive join using Shaw's approach rej-fb

Recursive join using a filter-based approach can simply be ignored from the join processing workflow, a fact that calls for their early elimination. We distinguish the following types of joins:

• Two-way join. Given two datasets R and L, a two-way join denotes the pairs of tuples r ∈ R and l ∈ L, such that r.k 1 = l.k 2 where k 1 and k 2 are join columns in R and L, respectively. The standard notation is: R k1=k2 L Notation: In order to simplify notations, we will often assume that join keys are known from the context, and will use the abbreviated form R L. • Multi-way join [START_REF] Tan | A Note on the Strategy Space of Multiway Join Query Optimization Problem in Parallel Systems[END_REF]. Given n datasets R 1 , R 2 , . . . , R n , we define a multiway join as a pairwise combination of two-way joins:

R 1 R 2 R 3 . . . R n
Considering only pairwise combination is a restriction: this subclass is sometimes called a chain join in the literature. • Recursive join [START_REF] Idreos | Continuous multi-way joins over distributed hash tables[END_REF][START_REF] Ordonez | Optimizing recursive queries in SQL[END_REF]. Given a relation K(x, y) encoding a graph, a recursive join computes the transitive closure of K. It requires an initialization, and an iteration (until a fixpoint occurs):

(Initialization) A(x, y) = K(x, y) (Iteration) A(x, y) = A(x, z) K(z, y)

We use the following running example: a user dataset R(uid, uname, location), a log dataset L(uid, event, logtime) and an acquaintance dataset K(uid1, uid2). These datasets illustrate the following searches.

• Q 1 -Two-way join. Find the names and events of all users who logged an event before 19/06/2015.

A 1 (uname, event) = π uname,event (R σ logtime<19/06/2015 (L))

• Q 2 -Multi-way join. Find the log events of all users known by Cang A 2 (uid, event, logtime) = π L (σ uname= Cang (R) uid=uid1 K uid2=uid L)

• Q 3 -Recursive join. List the ids of all connected to Philippe.

(

Initialization) A 3 (id) = π uid (σ uname= P hilippe (R)) (Iteration) A 3 (id) = π uid2 (K uid1=id A 3 )

MapReduce

MapReduce [START_REF] Dean | MapReduce: Simplified Data Processing on Large Clusters[END_REF] is a parallel and distributed programming model apt at running on computer clusters that scale to thousands of nodes in a fault-tolerant manner.

MapReduce usage has become widespread since Google first introduced it in 2004. It allows users to concentrate only on designing their data operations regardless of the distributed aspects of the execution.

(k 1 , v 1 ) map ---→ list(k2, v2)
The intermediate values associated with the same key k 2 are grouped by the framework and then passed to the reduce function which aggregates the values.

(k 2 , list(v 2 )) reduce ----→ v 3
As illustrated by Fig. 1, a typical MapReduce job is executed across multiple nodes. During the map phase, each map task reads a subset (called "split") of one input dataset, and applies the map function for each key/value pair of the split. The framework takes care of grouping intermediate data and sends them to the reducer nodes, a communication-intensive process called shuffling. Each reduce task collects the intermediate key/value pairs from all the map tasks, sorts/merges the data with the same key, and calls the reduce function to generate the final results.

MapReduce is designed to process a single dataset. Combining several inputs with a MapReduce framework is intricate. The problem has mostly been studied for joins.

Bloom filters

A Bloom filter (BF) [START_REF] Bloom | Space/Time Trade-offs in Hash Coding with Allowable Errors[END_REF] is a space-efficient randomized data structure used for testing membership in a set with a small rate of false positives.

A variant of a Bloom filter is Intersection Bloom filter [START_REF] Phan | Toward intersection filter-based optimization for joins in mapreduce[END_REF], denoted IBF (S 1 , S 2 ), is a probabilistic data structure designed to represent the intersection of sets S 1 and S 2 , and check membership in the intersection set. To achieve this, it computes the intersection of the Bloom filters BF (S 1 ) and BF (S 2 ). In join processing, matching a join key v against the intersection filter allows to decide (up to the false positive probability) whether it belongs to the shared join keys. The false positive probability of the intersection filter is estimated as f I representing one of the probabilities of different approaches to the filter [START_REF] Phan | Toward intersection filter-based optimization for joins in mapreduce[END_REF].

Extended Intersection filter [START_REF] Phan | Toward intersection filter-based optimization for joins in mapreduce[END_REF] (EIF ) is developed from the intersection Bloom filter. The EIF is a filter built on join key columns k 1 , k 2 , . . . , k m of datasets R 1 , R 2 , . . . , R m . It consists of Bloom filters hashed on the key columns,

BF 1 (R 1 .k 1 ), BF 2 (R 2 .k 2 ∩ R 3 .k 2 ), . . . , BF m (R m .k m ).
The membership test takes a tuple t(k 1 , k 2 , . . . , k m , . . . , k n ) and returns a "yes" or "no" answer indicating whether t is / is not in the filter. If one of the join keys of the tuple t, t(k i ) i=1...m , is not a member of the component filter BF i of the EIF , the output is "no" answer. Otherwise, the output is "yes" answer. Fig. 2 depicts its structure. For example, consider the three-way join R(uname, uid) K(uid1, uid2) L(uid, event). K can be filtered by an EIF composed of BF 1 (R.uid ∩ K.uid1) and BF 2 (K.uid2 ∩ L.uid), i.e., IBF 1 (R.uid, K.uid1) and IBF 2 (K.uid2, L.uid). Each tuple t(k 1 , k 2 ) ∈ K is checked against the two filters. If k 1 and k 2 are in IBF 1 and IBF 2 , respectively, t is accepted, else it is eliminated.

Joins with MapReduce

Join processing in MapReduce has become a hot research topic in recent years [START_REF] Blanas | A comparison of join algorithms for log processing in mapreduce[END_REF][START_REF] Afrati | Optimizing joins in a map-reduce environment[END_REF][START_REF] Hassan | Semi-join computation on distributed file systems using map-reduce-merge model[END_REF][START_REF] Afrati | Map-reduce extensions and recursive queries[END_REF][START_REF] Lee | Join processing using Bloom filter in MapReduce[END_REF][START_REF] Phan | Toward intersection filter-based optimization for joins in mapreduce[END_REF][START_REF] Bruno | Advanced join strategies for large-scale distributed computation[END_REF]. Many studies have been carried out to evaluate join queries and analyze large datasets in a MapReduce environment. Although joins in MapReduce can be implemented in many ways, the relative performance of the various algorithms depends on certain assumptions such as the size of inputs, data constraints, and joining rates. Map-side joins [START_REF] Blanas | A comparison of join algorithms for log processing in mapreduce[END_REF][START_REF] White | Hadoop: The Definitive Guide[END_REF][START_REF] Lee | Parallel data processing with mapreduce: A survey[END_REF] would be better to perform the entire joining operation in the map phase since it may save the shuffle and reduce phases. But this solution is limited in running extra MapReduce jobs to repartition the data sources to be usable. Meanwhile, Reduce-side joins [START_REF] Blanas | A comparison of join algorithms for log processing in mapreduce[END_REF][START_REF] White | Hadoop: The Definitive Guide[END_REF][START_REF] Lee | Parallel data processing with mapreduce: A survey[END_REF][START_REF] Liu | Efficient social network data query processing on MapReduce[END_REF] are more flexible and general to process a join operation as a standard MapReduce job without any constraints, but they are quite inefficient solutions. Joining does not take place until the reduce phase. In addition, the shuffle phase is really expensive since it needs to shuffle all data, sort and merge.

Observing Reduce-side joins shows that many intermediate pairs generated in the map phase may not actually participate in the joining process due to no matching with any pairs in another input dataset. Consequently, it would be much more efficient if we eliminate the non-matching data right in the map phase. This problem can be solved by Semi-join [START_REF] Blanas | A comparison of join algorithms for log processing in mapreduce[END_REF]. It uses a distributed cache to disseminate a hashmap of one of input datasets across all the mappers, then dropping tuples whose join key not in the hashmap. The main obstacle in this way resides at the hashmap because the hashmap may not fit in memory and its replication across all the mappers may be inefficient. In this situation, therefore, Bloom join [START_REF] Lam | Hadoop in Action[END_REF][START_REF] Lee | Join processing using Bloom filter in MapReduce[END_REF][START_REF] Zhang | Optimizing Distributed Joins with Bloom Filters Using MapReduce[END_REF][START_REF] Lee | Exploiting bloom filters for efficient joins in MapReduce[END_REF][START_REF] Zhang | Efficient processing distributed joins with bloom filter using mapreduce[END_REF] is a worthy replacement for Semi-join because it benefits from a Bloom filter [START_REF] Bloom | Space/Time Trade-offs in Hash Coding with Allowable Errors[END_REF] to do existence tests in less memory than a full list of keys from the hashmap. Another restriction on these solutions is derived from their filtering efficiency, even for recent research efforts [START_REF] Afrati | Optimizing joins in a map-reduce environment[END_REF][START_REF] Lee | Join processing using Bloom filter in MapReduce[END_REF][START_REF] Zhang | Efficient processing distributed joins with bloom filter using mapreduce[END_REF]. There remain a lot of non-matching data after filtering because the solutions can only filter on one of input datasets instead of both. Thus, Intersection filter-based join [START_REF] Phan | Toward intersection filter-based optimization for joins in mapreduce[END_REF] may become a better solution to address this problem by eliminating non-matching data from both input datasets. However, it is necessary to have a complete evaluation of the solutions that indicates their benefits and limitations.

In addition to the above two-way joins, the researchers are also confronted big challenges that come from multi-way joins and recursive joins in MapReduce. The multi-way join extends the two-way join by handling multiple input datasets, whereas the recursive join represents a computation of a repeated join operation. Both of them are still open issues and their existing solutions from traditional distributed and parallel databases cannot be easily extended to adapt to a shared-nothing distributed computing paradigm as MapReduce. For latest approaches, computing multi-way joins [START_REF] Afrati | Optimizing joins in a map-reduce environment[END_REF][START_REF] Blanas | A comparison of join algorithms for log processing in mapreduce[END_REF][START_REF] Zhang | Efficient processing distributed joins with bloom filter using mapreduce[END_REF][START_REF] Lee | Application of Filters to Multiway Joins in MapReduce[END_REF] and recursive joins [START_REF] Afrati | Map-reduce extensions and recursive queries[END_REF][START_REF] Afrati | Cluster computing, recursion and datalog[END_REF][START_REF] Bu | The HaLoop approach to large-scale iterative data analysis[END_REF][START_REF] Shaw | Optimizing Large-Scale Semi-Naive Datalog Evaluation in Hadoop[END_REF] also often generates intermediate results that may be inputs of component joins of the joins. These intermediate results contain a lot of non-matching data that considerably increases total overheads for I/O, CPU, sort, merge, and especially communication. We need to figure out optimized solutions that can prevent the non-matching data involved in the intermediate results. Besides, minimizing the intermediate data amount sent to the reducers should be calculated appropriately.

The purpose of the present paper is to provide a consistent review of filterbased join processing techniques in a MapReduce environment. It not only (a) covers the recently various techniques for computing two-way joins, multi-way joins and recursive joins, but also (b) qualifies these techniques with cost models and (c) evaluates them with experimental studies to both validate the proposed cost model and investigate their practical behavior. Overall, our goal is to provide a clear, robust and comparative assessment of join processing solutions to guide the choice of practitioners confronted to the need to perform join at scale in a specific context. By founding this assessment on both an analytic and empirical study, we hope to provide a material that puts the research contributions in this field in a coherent setting and clarifies the stakes of combining several inputs with MapReduce.

For the sake of consistency, we focus on join algorithms that share some common features. First, we only consider equijoins. Second, we investigate algorithms that exploit filters to reduce the network communication. Filtering is a strategy that can be combined with all kind of approaches, and turns out to be (almost) always beneficial in a context where I/O and network exchanges constitute the major bottleneck. Third, our work complements a few other surveys recently published [START_REF] Sakr | A survey of large scale data management approaches in cloud[END_REF][START_REF] Lee | Parallel data processing with mapreduce: A survey[END_REF][START_REF] Zhang | Efficient processing distributed joins with bloom filter using mapreduce[END_REF][START_REF] Sakr | The family of mapreduce and large-scale data processing systems[END_REF][START_REF] Doulkeridis | A survey of large-scale analytical query processing in mapreduce[END_REF][START_REF] Li | Distributed data management using MapReduce[END_REF] which, on the one hand, explore a larger scope (e.g., non-equi joins [START_REF] Okcan | Processing theta-joins using mapreduce[END_REF][START_REF] Zhang | Efficient multi-way theta-join processing using mapreduce[END_REF][START_REF] Zhang | Sej: An even approach to multiway theta-joins using mapreduce[END_REF][START_REF] Doulkeridis | A survey of large-scale analytical query processing in mapreduce[END_REF]), but on the other hand do not propose an in-depth coverage as we do, and a comparison methods uniformly applied to the range of proposals published so far.

Extending Equijoins with Filters in MapReduce

The most straightforward way to join datasets with MapReduce is the Reduceside join algorithm [START_REF] Blanas | A comparison of join algorithms for log processing in mapreduce[END_REF][START_REF] White | Hadoop: The Definitive Guide[END_REF][START_REF] Lee | Parallel data processing with mapreduce: A survey[END_REF][START_REF] Liu | Efficient social network data query processing on MapReduce[END_REF], denoted rsj. It groups tuples from both datasets on their respective join key value during the map phase, and merges/joins them during the reduce phase. Tuples are processed regardless of their actual contribution to the final result, and thus the join algorithm has to pay an overhead for processing and shipping useless data.

Consider for instance the Facebook user dataset R containing more than 1.23 billion users [START_REF]Facebook: Facebook reports fourth[END_REF]. We would like to obtain users' activities in a certain period of time (e.g., one hour) by joining R and the log dataset L. Since L, over this period, contains the activities of only a few million unique users, most of the users in R are not represented, and rsj spends useless resources to access, process and transfer over the network the non-matching tuples of R.

Several filter-based extensions have been proposed to tackle the problem. Their common idea is to filter out the non-matching tuples from the input datasets during the map phase. A filter in this context is a compact data structure that supports fast membership tests. Filter-based joins require two stages:

• Stage1 (pre-processing). A filter F is built on a join key value set of one input dataset. For the intersection filter, F represents the intersection of the key value sets. A membership test for some key value k on F tells whether k participates or not to the join result. • Stage 2 (join). F is distributed to all the computing nodes, and used to eliminate non-matching tuples during the map phase. The join then proceeds as explained above.

A filter is a compact representation of a set. It accepts a rate of false positives (i.e., positive answer for non-matching tuples in our case) but no false negatives. Filtering avoids the communication overhead of shipping tuples from the mappers to the reducers, and the storage and CPU overhead of processing such tuples during the reduce phase. The join strategy remains unchanged, and exploits the MapReduce paradigm: the input datasets are partitioned and grouped during the map phase, in order to solve locally the problem during the reduce phase. Filtering presents some advantages and disadvantages:

• Advantages: the strategy does not impose any restrictions on input datasets, nor modifications to the MapReduce framework. Besides, it removes nonmatching data to reduce the communication overhead. • Disadvantages: building the filters represents a significant cost, since it requires scanning the input, and transferring the filters.

In the rest of this section, we examine in detail the application of filter-based techniques to the following join variants: two-way joins, multi-way joins and recursive joins. For each variant, we present the state-of-the-art algorithms, along with a discussion on their expected advantages/disadvantages.

Two-way joins

A two-way join R 1 R 2 involves two relations R 1 and R 2 . In the following r 1 (resp. r 2 ) denote a tuple from R 1 (resp. R 2 ) and k refers to the join key attribute. We use simplified notations when allowed by the context.

Bloom joins.

Bloom join (bj) [START_REF] Lam | Hadoop in Action[END_REF][START_REF] Lee | Join processing using Bloom filter in MapReduce[END_REF][START_REF] Zhang | Optimizing Distributed Joins with Bloom Filters Using MapReduce[END_REF] is a specific type of the filter-based join strategy in which the well-known Bloom filter [START_REF] Bloom | Space/Time Trade-offs in Hash Coding with Allowable Errors[END_REF] is used. bj is implemented by two MapReduce jobs as follows:

• Job 1 (preprocessing) is a job with only one reducer. The mappers scan splits of the input R 2 , extract the join key value from each tuple, and produces local Bloom filters. Then, the mappers emit the local filters to the reducer that merges them into a global filter BF (R 2 ) using the bit-wise OR.

• Job 2 (processing) filters out non-matching tuples in R 1 and joins the filtered result R 1 with R 2 . It relies on a distributed cache to store BF (R 2 ). The mappers scan splits of R 1 and R 2 , and eliminate the tuples of R 1 whose keys are not in BF (R 2 ). Tuples from R 2 are not filtered. Each tuple is then ticked with a tag that indicates its dataset name.

For our example, mappers emit tagged tuples with composite keys of the form ((r 1 .k, 'R 1 '), r 1 ) or ((r 2 .k, 'R 2 '), r 2 ). The reducers receive tagged tuples grouped on the k value (this requires a small change of the partitioning function). For each group, the reduce function constructs all the pairs (r 1 , r 2 ) to complete the join.

Note that it requires to override the default grouping function in order to ensure that grouping the tagged tuples takes into consideration only the join key part and ignores the tag part. The tag is used for secondary sort which ensures that, for a given key value, all tuples from R 1 are processed before those of R 2 . This allows to apply a standard in-memory hash join.

Discussion. bj benefits from the compacity of the Bloom filter to reduce the amount of data transferred over the network. The size of the filter can be fixed regardless of the number of join keys. However, given a fixed filter size, the probability f of false positives increases with the number of join keys.

A major concern with the filtering approach in general is the need to run a pre-processing job for building the filter. Besides, broadcasting the filter becomes inefficient if its size is large. Finally, it is worth noting that the bj is asymmetric: non-matching tuples of R 2 have not been filtered, hence the problem is halfsolved.

The authors of [START_REF] Lee | Join processing using Bloom filter in MapReduce[END_REF] have proposed an improvement of bj that avoids the preprocessing job, but requires two internal modifications of the framework. We do not consider in the present study such extensions that necessitate a non-standard environment.

Intersection filter-based joins.

We now describe an improvement of the above approach, the Intersection filter-based join [START_REF] Phan | Toward intersection filter-based optimization for joins in mapreduce[END_REF], denoted ifbj. It relies on the fact that only tuples whose join keys belong to the set of shared join keys do participate to the result.

The implementation of ifbj is done with the following jobs:

• Job 1 (pre-processing) is a job with only one reducer. The mappers scan splits of R 1 and R 2 , extract the join key value for each tuple, and insert them in the local Bloom filters regardless of duplicate keys. The mappers then emit the local filters to the reducer which merges them in two global filters BF (R 1 ) and BF (R 2 ) using the bit-wise OR. Based on one of three approaches to building the intersection filter [START_REF] Phan | Toward intersection filter-based optimization for joins in mapreduce[END_REF], the reducer computes the intersection filter IBF (R 1 , R 2 ) from the global filters. • Job 2 (join) uses a distributed cache to provide IBF to all the compute nodes. The mappers scan splits of R 1 and R 2 , extract the join key for each tuple and match it against the intersection filter. If the key v belongs to the intersection filter, the tuple is emitted as a pair ((v, tag), tuple). The join evaluation in the reduce phase is similar to the Bloom join algorithm.

ifbj benefits from the standard features of Bloom filters: its small size, its independence from the number of the keys and key duplication, and fast membership test. Join based on the intersection filter is expected to be more efficient than the Bloom join because of its ability to filter out non-matching tuples from both two input datasets. An interesting characteristic of the intersection filter is that if IBF (R 1 , R 2 ) has all bits set to zero, then the sets R 1 .k and R 2 .k are disjoint and the join evaluation stops without doing anything. However, the algorithm has to pay the additional cost of a MapReduce job for building the intersection filter and requires scanning the two input datasets twice.

Multi-way joins

We can extend the above approach to the computation of multi-way joins with an extended intersection filter (EIF ) in the following.

Three-way joins.

We begin our study of multi-way joins by considering the special case of a three-way join R 1 R 2 R 3 . For the sake of concreteness, we will discuss the following query on our example relations.

R uid=uid1 K uid2=uid L
There are several possible pairwise combinations to compute this three-way join. We can evaluate three-way joins as a sequence of 2 two-way joins, using two successive jobs. An alternative is to join the three datasets together with a single job, as recently proposed by Afrati and Ullman [START_REF] Afrati | Optimizing joins in a map-reduce environment[END_REF]. It relies on the idea of a matrix of reducers as shown in Fig. 3.

R uid=uid1 K uid2=uid L = (R uid=uid1 K) uid2=uid L = R uid=uid1 (K uid2=uid L)
The number of reducers must be the square of some integer n (r = n 2 ) and reducers are mapped (virtually) to a matrix n × n. Each reducer is mapped to a cell (i, j), and identified by i * n + j. With n = 5, cell (3, 2) is for instance associated with the reducer 17.

The mappers assign tuples of R, K, and L to the reducer matrix as follows. Let h be a hash function with range [0, n -1]. Each tuple of K is sent to a single reducer, the one in cell (h(K.uid1 ), h(K.uid2 )). Tuples from R and L are sent to all the reducers of, resp. a whole row or column in the matrix. Each tuple r(uid, uname) is sent to all the reducers of the row h(uid). Each tuple l(uid, event) is sent to all the reducers of the column h(uid).

We can give a perspective: assume three tuples R('Laurent', u 1 ), K (u 1 , u 2 ), and L(u 2 , 'login'). They will all be sent to the reducer h(u 1 ) * n + h(u 2 ) and the joined tuple will therefore be produced.

Let us assume, for simplicity, that |R|=|K |=|L|. The total communication cost for the Afrati's three-way join (denoted 3wj in the following) is O(|R|.

√ r), whereas the total communication cost for the cascade of 2 two-way joins without filters is O(|R| 2 .α), where α is the probability for two tuples from different datasets to match on the join key (Section 4.2 for more details). It follows that 3wj is better than the cascade of the two-way joins when r < (|R|.α) 2 .

A downside of 3wj is that it generates n duplicates for each tuple of either R or L. This represents a large communication and I/O overhead. This situation can be improved significantly by removing non-matching tuples prior to the reduce phase. We extend 3wj with intersection filters as shown in Fig. 4. The extension of the three-way join with filters uses two jobs as follows.

• Job 1 (pre-processing) builds IBF 1 (R.uid, K.uid1) and IBF 2 (K.uid2, L.uid).

Let mp 1 , mp 2 and mp 3 be the number of map tasks for R, K and L, respectively. The job consists of mp 1 + mp 2 + mp 3 map tasks that build filters and one reduce task that produces two intersection filters. In detail, mp 1 tasks build local filters on R.uid; mp 2 tasks build local filters on K.uid1 and K.uid2; mp 3 tasks build local filters on L.uid. Those filters are shipped to the reducer which produces BF (R.uid), BF (L.uid), BF (K.uid1), BF (K.uid2), as well as IBF 1 , IBF 2 and EIF (IBF 1 , IBF 2 ). Note that the join result is known to be empty right away if either IBF 1 or IBF 2 is empty. • Job 2 (join) filters out non-matching tuples from R, K and L, and carries out the join evaluation. It distributes the intersection filters to all tasktrackers, creates map tasks for R, K and L and r reduce tasks.

Map phase with filtering: Each mapper matches any tuple of R or L against the relevant filter IBF 1 , IBF 2 , or EIF (IBF 1 , IBF 2 ). Tuples that pass the filtering process are then sent to the reducers according the 3wj policy. This involves tuple replication as shown in Fig. 3.

Reduce phase: the reduce function applies a full cross-product of tuples from the different input datasets. Locally, the reducer buffers the tuples of R and L, streams the tuples from K, and performs the cross product.

Chain joins.

We now consider the more general case of multi-way joins, or chain joins, a sequence or pair wise joins of the form of

R 1 (x 1 , x 2 ) R 2 (x 2 , x 3 ) R 3 (x 3 , x 4 ) • • • R n (x n , x n+1 ).
The baseline solution is a cascade of Bloom joins (cj-bj). The query plan is a left-deep join tree, and relies on a set of filters BF 2 (R 2 .x 2 ), . . . , BF n (R n .x n ) built on the base datasets by a pre-processing job. In this scenario, we can regconize that R 1 and all intermediate results We propose an improved evaluation that generalizes intersection filters as shown by Fig. 5. In addition to the filters BF on base relations, the extended algorithm denoted cj-ifbj creates on the fly intersection Bloom Filters on intermediate results,

IBF (R 1,••• ,i-1 .x i , R i .x i ), i ∈ [2, n]
during the reduce phases of intermediate joins All the input datasets and intermediate join results are filtered by their corresponding intersection filters. For instance, IBF (R 1,2 .x 3 , R 3 .x 3 ) is used to eliminate non-matching tuples in both R 1,2 and R 3 . Intermediate data sent to the reducers with cj-ifbj is expected to be much less than in the case of cj-bj.

We can even go one step further by noting that intermediate join results still contain non-matching tuples transmitted to the next join. For instance, the join of R 1 and R 2 likely contains tuples that do not match any tuples of R 3 on x 3 . We can therefore "push" the filter BF (R 3 .x 3 ), down to the scan of relation R 2 . The idea is actually quite reminiscent of the traditional optimization heuristics that pushes selection down the query tree in relational systems.

(R 1,2,••• ,n-1 .
x n ). This solution ensures that intermediate join results only contain (up to false positives) matching data that can be sent to the next join without filtering. This is an important characteristic which avoids to apply additional filters to intermediate join results.

The implementation first uses a pre-processing job to build the Bloom filters BF (R i .x i ), i = 2, . . . , n, and BF (R 1 .x 2 ). Next, it evaluates the chain join as a sequence of two-way joins. During the evaluation of

R 1,••• ,i-1 R i , the left input need not be filtered, except R 1 filtered by BF (R 2 .x 2 ). The right input is filtered by the EIF i built from BF (R 1,••• ,i-1 .x i ) and BF (R i+1 .x i+1
). The former is generated in the reduce phase of the previous join processing between R 1,••• ,i-2 and R i-1 . Building the filters from the intermediate join results does not involve any overhead. The iteration stops if one of the two input datasets is null.

Fig. 6 (b) illustrates a second optimization, where pairwise joins are replaced by filtered three-way joins (3wj). We denote this further optimized solution as ocj-3wj. Consider the three-way join

R 1,••• ,i-1 R i R i+1 , i ∈ [2, n -1]
and i is an even number. The left relation does not need to be filtered, apart from R 1 filtered by BF (R 2 .x 2 ). The middle relation is filtered by the extended intersection filter

EIF i built from BF (R 1,••• ,i-1 .x i ) and a fil- ter IBF (R i .x i+1 , R i+1 .x i+1 ). The last input is filtered by EIF i , built from IBF (R i .x i+1 , R i+1 .x i+1 ) and BF (R i+2 .x i+2 ). When (i + 2) > n, the filter EIF i does not contain BF (R i+2 .x i+2 ) because R i+2 does not exist. It is noted that ocj-3wj may contain a two-way join of R 1,••• ,n-1 and R n if n is an even number.
The implementation of the second solution is similar to the first one. ocj-2wj is expected to use less memory than ocj-3wj because the former only buffers one input for each two-way join, whereas the second one must buffer two inputs for each three-way join. The downside is that ocj-2wj requires more jobs than ocj-3wj. If n denotes the number of input datasets, the number of the two-way join jobs of ocj-2wj is (n -1), while ocj-3wj needs n-1 2 jobs.

Recursive joins

We now turn to another complex type of join. A recursive join [START_REF] Idreos | Continuous multi-way joins over distributed hash tables[END_REF][START_REF] Ordonez | Optimizing recursive queries in SQL[END_REF] computes the transitive closure of a relation encoding a graph. A typical example, expressed in Datalog, is given below.

F riend(x, y) ←-Know(x, y) F riend(x, y) ←-F riend(x, z) Know(z, y)
Evaluating a recursive join is tantamount to computing the transitive closure of the graph represented by the relation. This can be done via an iterative process that stops whenever a fixpoint is reached. We examine how the seminaive algorithm [START_REF] Ullman | Principles of Database and Knowledge-Base Systems[END_REF] can be evaluated in MapReduce.

Let F and K denote the relations Friend and Know, respectively. Let F i , 0 < i ≤ n be the temporary value of the relation Friend at step 0, with F 0 = ∅. The incremental relation of F i , i > 0, denoted ∆F i , is defined as:

∆F i = F i -F i-1 = Π xy (∆F i-1 z K) -F i-1
The semi-naive algorithm uses this delta relation to avoid redundant computations (Algorithm 1).

Algorithm 1: Semi-Naive evaluation for recursive joins

Input: A graph encoded as a relation K Output: The transitive closure of K 1 F = ∅, ∆F0 = K(x, y), i = 1 2 while ∆Fi-1 ! = ∅ do 3 ∆Fi = Πxy(∆Fi-1 z K) -F 4 F = F ∪ ∆Fi 5 i+ = 1 6 return F
At each step i, some new facts are inferred and stored in ∆F i . The loop is repeated until no new fact is inferred (∆F i = ∅), i.e., the fixpoint is reached. The union of all incremental relations, (∆F 0 ∪ . . . ∪ ∆F i-1 ), is the transitive closure of the graph.

Shaw et al. [START_REF] Shaw | Optimizing Large-Scale Semi-Naive Datalog Evaluation in Hadoop[END_REF] have proposed the following algorithm to implement the semi-naive algorithm in MapReduce (rej-shaw). Each iteration evaluates ∆F i = Π xy (∆F i-1 z K) -F i-1 with two jobs, namely, one for join and one for deduplication and difference (dedup-diff ), as shown on Fig. 7. The first job computes (∆F i-1 K), the second computes the new delta relation ∆F i . This 2-jobs execution is iterated until ∆F i is empty. This means that the invariant relation K and the incremental relation F i-1 are re-scanned and re-shuffled for every iteration. Shaw et al. have tackled this situation in the HaLoop system [START_REF] Bu | The HaLoop approach to large-scale iterative data analysis[END_REF] by using the Reducer Input Cache (RIC). RIC stores and indexes reducer inputs across all reducers. To avoid re-scanning and re-shuffling the same data with the same mapper on iterations, the solution therefore uses RIC for the datasets K and F i-1 in the join job and the dedup-diff job, respectively, as shown on Fig. 7. K is scanned only once, at the first loop. K i and K j are splits of K, which are cached at the reducer input caches i and j, resp. Note that caching intermediate results during iterative computations is now integrated in modern distributed engines such as Spark [7] and Flink [START_REF]Apache: Flink[END_REF].

The dedup-diff job using RIC is described as follows. Each tuple is stored in the cache as a key/value pair (t, i ), where the key is the tuple t discovered by the previous join job and the value is the iteration number i for which that tuple was discovered. The map phase of the difference job hashes the incoming tuples as keys with values indicating the current iteration number. During the reduce phase, for each incoming tuple, the cache is probed to find all instances of the tuples previously discovered across all iterations. Both the incoming and cached data are passed to the user-defined reduce function. A tuple previously discovered is omitted from the output, else it is included in ∆F i .

When evaluating (∆F i-1 K), Shaw's solution (rej-shaw) does not discover and eliminate non-matching tuples from ∆F i-1 and K. Our extension, rej-fb in the following, adds an intersection filter IBF (∆F i-1 .z, K.z) as proposed in Section 2.3. Initially, the filter is simply BF (K.z) generated by a preprocessing job. During the i th iteration (i ≥ 1), rej-fb uses IBF (∆F i-1 .z, K.z) as a filter in the map phase of the join job, and builds IBF (∆F i .z, K.z) in the reduce phase of the dedup-diff job.

A fixpoint of the recursive join is reached when no new tuples are discovered (i.e. ∆F i is empty) or, equivalently, when the IBF is empty. The latter is a better stop condition because it can save one iteration.

Performance analysis for filter-based equijoins

We now develop an analysis of the algorithms presented so far.

Two-way joins

We note R and L the two input datasets, and analyze the cost for, respectively, the Bloom join (bj) and the intersection filter-based join (ifbj). Table 2 summarizes the parameters of our cost model.

Cost model.

We adapt the cost model presented in [START_REF] Nykiel | MRShare: Sharing Across Multiple Queries in MapReduce[END_REF]. We propose the following global formula that captures the cost of a two-way join.

C 2wJoin = C pre + C read + C sort + C tr + C write (1)
where:

• C read = c r • |R| + c r • |L| • C sort = c l • |D| • 2 • ( log B |D| -log B (mp) + log B (mp) ) [26] • C tr = c t • |D| • C write = c r • |O| • C pre = C + c r • m • t C = C read + (c l + c t ) • m • mp , for ifbj c r • |L| + (c l + c t ) • m • mp 2
, for bj C pre = 0, for approaches without filters. In addition, it is assumed that the filters are the same size m. If m is small, we will not compress the filter files and m is therefore the size of the Bloom filter. The cost of reading or writing data locally cr

The cost of reading/writing data remotely ct

The cost of transferring data from one node to another B +1

The size of the sort buffer in pages mp1

The number of map tasks for R mp2

The number of map tasks for L mp

The total number of map tasks, mp = mp1 + mp2 t

The number of tasktrackers m

The compressed size of the Bloom filter (bits) m = the size of the Bloom filter × the file compression ratio

|O|

The size of the join processing output Cpre

The total cost to perform the pre-processing job C read

The total cost to read the data Csort

The total cost to perform the sorting and copying at the map and reduce nodes

Ctr

The total cost to transfer intermediate data among the nodes

Cwrite

The total cost to write the data on DFS An additional component, C pre , is added to the cost model in [START_REF] Nykiel | MRShare: Sharing Across Multiple Queries in MapReduce[END_REF] to form equation [START_REF] Afrati | Cluster computing, recursion and datalog[END_REF]. |D|, the size of the intermediate data, strongly influences the total cost of a join operation, and is essential in particular to decide whether the filter-based variant of the algorithm is worth its cost.

Cost comparison.

In this section, we evaluate |D|, for each algorithm mentioned in Section 3.1, and provide a cost comparison. Importantly, we identify a threshold that can guide the choice amongst of these algorithms. We add the Reduce-side join (rsj) to our comparison to highlight the effect of filtering.

We denote as δ L and δ R , respectively, the ratio of the joined records of R with L (resp. L with R). The size of intermediate data is:

|D| =      δ L |R| + f I (R, L) • (1 -δ L )|R| + δ R |L| + f I (R, L) • (1 -δ R )|L| (2) δ L |R| + f (L) • (1 -δ L )|R| + |L| (3) |R| + |L| (4)
where:

• equation ( 2 From these equations, we can infer the following. Theorem 1. An ifbj is more efficient than a bj because it produces less intermediate data. Additionally, the following inequality holds:

D ifbj ≤ D bj ≤ D rsj (5)
where D ifbj , D bj , and D rsj are the sizes of intermediate data of ifbj, bj, and rsj, resp.

Proof. We get 0 < f I (R, L) < f (L) < 1 [START_REF] Phan | Toward intersection filter-based optimization for joins in mapreduce[END_REF]. So we can deduce that:

δ L • |R| + f I (R, L) • (1 -δ L ) • |R| ≤ δ L • |R| + f (L) • (1 -δ L ) • |R| ≤ |R| and (6) δ R • |L| + f I (R, L) • (1 -δ R )|L| ≤ |L| (7)
Note that the ratio of the joined records, δ L or δ R , could be 1 in the case of a join based on a foreign key.

By combining inequalities ( 6) and (7) into equations ( 2), ( 3) and ( 4), Theorem 1 is proved .

From equations (1) and ( 5), we can evaluate the total cost of the join operation for the different approaches.

Theorem 2. Once the pre-processing cost C pre is negligible or less than the cost of non-matching data, an ifbj has the lowest cost. In addition, a comparison of the costs is given by:

C ifbj ≤ C bj ≤ C rsj (8)
where C ifbj , C bj , and C rsj are the total costs of ifbj, bj, and rsj, resp. As a result, the most efficient join approach is typically ifbj, the second one is bj, and the worst one is rsj.

The total cost to perform the pre-processing job is given by:

C pre =      C read + (c l + c t ) • m • mp + c r • m • t , in case of ifbj c r • |L| + (c l + c t ) • m • mp 2 + c r • m • t , in case of bj 0 , in case of rsj
Regarding data locality, the MapReduce framework makes its best efforts to run the map task on a node where the input data resides. Although this cannot always be achieved, we can see that the cost of this phase, C pre , is negligible compared to the generation and transfer of non-matching tuples over the network. In general, choosing the filter-based joins relies on the read cost c r and a threshold of non-matching data shown in Theorem 3.

The filter-based join algorithms will become inefficient when there is a large number of map tasks, and very little non-matching data in the join operation. For large inputs with many map tasks, a tasktracker running multiple map tasks will maintain only two local filters BF (R) and (or) BF (L) thanks to merging the local filters of the tasks. Two hundred map tasks running on a tasktracker, for instance, will produce 200 local filters BF (R). The tasktracker merges all the local filters into one BF (R). Besides, as the number of non-matching tuples decreases, the filters become useless and computing them with an additional job represents a penalty. It hence needs to indicate the dependence of the filter-based joins on the amount of non-matching data through estimating the threshold for this data that determines whether filters should be used.

Let |D * | be the size of non-matching data, C * sort be the total cost of sorting and copying it at the map and reduce nodes, and C * tr be the total cost to transfer it among the nodes. Accordingly, the cost associated with non-matching data is the sum of C * sort and C * tr .

Theorem 3. The filter-based joins become a good choice when:

C pre < C * sort + C * tr ( 9
)
where:

• |D * | = |R| + |L| -|D| • C * tr = c t • |D * | • C * sort = c l • |D * | • 2 • ( log B |D * | -log B (mp) + log B (mp) ) [26]
Based on the size of |D|, the threshold depends on δ L and δ R (the ratio of the joined records).

In summary, the best choice of the join approaches is ifbj, the second one is bj, and the worst one is rsj (Theorem 2). However, this would become incorrect when the join has small input datasets and a high ratio of matching tuples that is defined by the threshold of the joined records (Theorem 3). In these cases, rsj would be the best choice and the filter-based joins should not be used because the cost of building and broadcasting filter(s) becomes relatively significant.

Multi-way joins

Three-way joins.

Let R, K and L be three input datasets. The general formula that estimates the total cost of 3wj is:

C 3wJoin = C pre + C read + C sort + C tr + C write (10)
where:

• C read = c r • |R| + c r • |K| + c r • |L| • C sort = c l • |D| • 2 • ( log B |D| -log B (mp) + log B (mp) ) [26]
• mp = mp 1 + mp 2 + mp 3 , the total number of map tasks for the three inputs

• C tr = c t • |D| • C write = c r • |O| • C pre = C read + (c l + c t ) • m • mp + 2 • c r • m • t, for 3wj using filters; C pre = 0 for 3wj.
To simplify the computation, we suppose that R, K and L have the same size. A 3wj increases the communication cost because each tuple of R and L is sent to many different reducers. On the other hand, the two-way join cascade must launch an additional job, then scan and shuffle the intermediate result. We characterize the relative costs of the approaches as follows.

|R| • c + |K| + |L| • b (11)
We must find optimal values for b and c to minimize the above expression subject to the constraint that b • c = r, b and c being positive integers. In this case, the Lagrangian multiplier method is used to present the solution.

Here

L = |R| • c + |K| + |L| • b -λ • (b • c -r).
We consider the problem min b,c≥0

[|R| • c + |K| + |L| • b -λ • (b • c -r)]
We make derivatives of L with respect to variables b and c.

∂L ∂b = |L| -λ • c = 0 ⇒ |L| = λ • c ; ∂L ∂c = |R| -λ • b = 0 ⇒ |R| = λ • b
We obtain the Lagrangian equations:

|L| • b = λ • r,
|R| • |L| • r/ |R| + |K| + |L| • |R| • r/ |L| ≈ 2 • |R| • √ r
Next, we specify the intermediate data size of the cascade of 2 two-way joins:

|R| • |K| • α + |L| ≈ |R| 2 • α (where |R| • α > 1)
The cost of 3wj, O(|R| • √ r), is compared with the cost of the two-way join cascade O(|R| 2 • α). We can conclude that 3wj will be better than the cascade when √ r < |R|•α. In other words, for 3wj, there is a limit on the number of reducers r < (|R|•α) 2 and Theorem 4 is hence proved .

In general, we can extend Theorem 4 for 3wj with n join key columns using an n-dimensional reducer matrix. For example, a 3wj R(A, B ) K (B, C ) L(C, A) with three join attributes A, B, and C. This three-way join needs a three-dimensional reducer matrix. The three-way join will become more efficient than a cascade of 2 two-way joins when r < (|R|.α) 3 and its amount of communication is 3

•|R|• 3 √ r.
In fact, choosing the number of reducers to satisfy this condition is not difficult. For example, if |R|•α = 15, as might be the case for the Web incidence matrix, we can choose the number of reducers r up to 3375. We can now characterize the cost of three-way join using filters.

Theorem 5. A 3wj, R(A, B) K(B, C) L(C, D)
, is more efficient with filters than without filters when C pre is negligible or less than the cost of processing non-matching data. Moreover, the 3wj using the filters is also more efficient than the two-way join cascade using the filters when r < (|R | • α) 2 . With using the filters, the size of the intermediate data is defined by

|D | = 2 • |R | • √ r , for 3wj using the filters. |R | 2 • α
, for a cascade of 2 two-way joins using the filters.

|R | = δ • |R| + f I • (1 -δ) • |R|, R is the filtered dataset of one input.
where r is the number of reducers, α is the probability of two tuples from different datasets to match on the join key, |R| = |K| = |L|, δ is the ratio of the joined records of one input dataset with another, and f I is the false intersection probability between the datasets.

Proof. Theorems 2 and 3 show that joins with the filters is more efficient than without the filters if C pre is negligible or less than the cost of non-matching data.

The following inequalities hold: 0 < δ << 1 and 0

< f I << 1 ⇒ δ • |R| + f I • (1 -δ) • |R| < |R| ⇒ |R | < |R| Combining this
equality with Theorem 4, we can easily prove Theorem 5 .

Chain joins.

Consider a chain join over n input datasets R 1 , R 2 , . . . , R n . We analyze ocj-3wj with the EIF filters presented in Section 3.2. The chain join is executed as a sequence of 3wj jobs, ... J = {J 2 , J 4 , J 6 , . . . , J n-1 }. J 1 scans R 1 , . . . , R n inputs for building the filters. Each iteration carries out the join of three inputs, R 1,••• ,2i-1 , R 2i , and R 2i+1 , where 1 ≤ i ≤ (n -1)/2 . If n is even, ocj-3wj contains an additional two-way join job of R 1,••• ,n-1 and R n . We extend the cost model of 3wj as follows: [START_REF] Bu | The HaLoop approach to large-scale iterative data analysis[END_REF] where:

C( ... J ) = C pre + (n -1)/2 • C distCache + C 2wJoin + (n-1)/2 i=1 ( C read (J 2i ) + C sort (J 2i ) + C tr (J 2i ) + C write (J 2i ) )
• C pre = ( n i=1 c r • |R i |) + (c l + c t ) • m • mp C pre =
0 and m = 0 for approaches without using filters. mp is the total number of map tasks.

• C distCache = 3 • c r • m • t C distCache =
0 for approaches without using filters.

• C 2wJoin is specified by equation ( 1), the cost of joining R 1,••• ,n-1 and R n .

C 2wJoin = 0 if n is an odd number and greater than 2.

• C read (J 2i ) = c r • |R 1,••• ,2i-1 | + c r • |R 2i | + c r • |R 2i+1 | • C sort (J 2i ) = c l • |D i | • 2 • ( log B |D i | -log B (mp) + log B (mp) ) [26] |D i | is the size of the intermediate data in the i th iteration. • C tr (J 2i ) = c t • |D i | • C write (J 2i ) = c r • |R 1,••• ,2i+1 | + a a = 2 • c r • m • t, for building BF (R 1,••• ,2i+1 ) in the i th iteration. a = 0, for (2i + 1) = n.
The computation of ocj-2wj is a sequence of (n-1) two-way join jobs. This computation can be also considered as a sequence of ((n-1)/2) three-way join jobs in which each of them is executed by a cascade of 2 two-way join jobs. As a result, ocj-2wj has the extra costs of writing and re-reading the intermediate results of the two-way joins, and initializing additional jobs. On the other hand, ocj-3wj incurs the costs of data duplication to the reducers. From Theorem 5, we can show that ocj-3wj is more efficient than ocj-2wj when r < (|R | • α) 2 .

Recursive joins

Cost model.

In the semi-naive algorithm, the number of iterations l is the longest path length in the relation graph minus 1, called the depth of the transitive closure. The first job J 1 reads K and ∆F 0 = F, and caches K at the reducers. Each subsequent job J i reads ∆F i-1 and scans partitions of K cached at the reducers (RIC). The dedup-diff job I i reads the join output O i containing duplicates, maps and shuffles tuples of O i to the reducers in order to generate ∆F i .

We base our analysis on the cost model introduced in [START_REF] Nykiel | MRShare: Sharing Across Multiple Queries in MapReduce[END_REF] and adapt it to the evaluation of the recursive join. Table 3 gives the parameters.

The total cost of the recursive join is specified by:

C( Ĵ) = C K + l i=1 C read (J i ) + C sort (J i ) + C tr (J i ) + C cache (J i ) + C write (J i ) + l i=1 C read (I i ) + C sort (I i ) + C tr (I i ) + C cache (I i ) + C write (I i ) (13)
where:

• C K = c r • |K| + c l • |K| • 2 • ( log B |K| -log B (mp K ) + log B (mp K ) ) + (c t + c l ) • |K| • C read (J i ) = c r • |∆F i-1 | • C sort (J i ) = c l • |D i | • 2 • ( log B |D i | -log B (mp ∆F i-1 ) + log B (mp ∆F i-1 ) ) [26] • C tr (J i ) = c t • |D i | • C cache (J i ) = c l • |K | • C write (J i ) = c r • |O i | • |D i |= |∆F i-1 |= β i-1 • |O i-1 | • C read (I i ) = c r • |O i | • C sort (I i ) = c l • |D + i | • 2 • ( log B |D + i | -log B (mp Oi ) + log B (mp Oi ) ) [26] • C tr (I i ) = c t • |D + i | • C cache (I i ) = c l • |D + i |• (|F i-1 | / r ) + c l • |∆F i | • (|F i-1 | / r + 1) • C write (I i ) = c r • |∆F i | • |D + i |= |O i | • |∆F i |= β i • |O i |
The average size of the cache at each reducer is (|F i-1 | / r ). For each incoming tuple of O i , the reducer probes the cache to get all tuples previously discovered. For each new tuple discovered, the reducer rewrites its entire cache along with the new tuple. Therefore, the total cost of accessing the cache in the dedup-diff job, C cache (I i ), includes the costs of reading the reducer cache for tuples of O i and rewriting the reducer cache for new tuples of ∆F i . The cost of reading or writing data locally cr

The cost of reading/writing data remotely ct

The cost of transferring data from one node to another B +1

The size of the sort buffer is B +1 pages (all costs are measured in seconds per page) mpK

The total number of map tasks of the dataset K mp∆F i-1

The total number of map tasks of the incremental relation ∆Fi-1 mpOi

The number of the map tasks of the join output (Oi) r

The number of reduce tasks t

The number of tasktrackers

|K |

The size of the dataset K that is invariant in loops |∆Fi-1|

The size of the incremental relation in the (i -1)

th iteration (|∆F0| = |K|). |∆Fi|
The size of the incremental relation in the i th iteration. The dataset ∆Fi contains only the differences between the join output Oi and Fi-1.

|Fi-1|

The size of all incremental relations in the iterations 0 to i -

1 (|∆F0 ∪ • • • ∪ ∆Fi-1|) |Di|
The intermediate data size of the join job Ji in the i th iteration.

|D + i|
The intermediate data size of the dup-diff job Ii in the i th iteration. |Oi|

The size of the join processing output Oi. The output Oi may contain duplicate elements with Fi-1 (previous incremental relations) βi

The difference ratio of the output Oi with Fi-1 CK

The total cost to read, map and sort, shuffle, and cache K at the reducers (RIC) in the first iteration C read (Ji)

The total cost to read the incremental relation ∆Fi-1 from DFS Csort(Ji)

The total cost to perform the sorting and copying of the join job at the map and reduce nodes Ctr(Ji)

The total cost to transfer intermediate data of the join job among nodes C cache (Ji)

The total cost to locally read partitions of K cached at the reducers Cwrite(Ji)

The total cost to write Oi to DFS C read (Ii)

The total cost to read the join output Oi from DFS Csort(Ii)

The total cost to perform the sorting and copying of the dup-diff job at the map and reduce nodes Ctr(Ii)

The total cost to transfer intermediate data of the dup-diff job (=Oi) among the nodes C cache (Ii)

The total cost to locally read partitions of Fi-1 cached at the reducers Cwrite(Ii)

The total cost to write ∆Fi to DFS.

Cost comparison. The total cost of rej-fb is smaller than that of rej-shaw because the intermediate data of rej-fb is less than that of rej-shaw

(|D i | < |D i |).
The amount of intermediate data of rej-fb is defined by:

|D i | = δ i-1 K • |∆F i-1 | + f (K) • (1 -δ i-1 K ) • |∆F i-1 | = δ i-1 K • |D i | + f (K) • (1 -δ i-1 K ) • |D i | < |D i | (14)
where:

• δ i-1 K
is the ratio of the joined records of ∆F i-1 with K • f (K) is the false positive probability of the Bloom filter BF (K.z)

We need a pre-processing job for building the Bloom filter BF (K.z ) that is used in all iterations. The additional overhead of building the filter BF (K.z ) is:

C K = C K + C pre (15)
where:

• C pre = c r • |K | + (c l + c t ) • m k • mp k + c r • m k • t • m k is the compressed size of the Bloom filter of the input dataset K (bits).
It is the product of the size of the filter and the file compression ratio. If the size of the filter is small, the file compression ratio should be one.

Besides, on each iteration, the program also re-computes the global filter BF (∆F i .z) generated in the reduce phase of the dedup-diff job. The overhead of creating the filter BF (∆F i .z) is:

C write (I i ) = C write (I i ) + (2 • c r • m ∆F i • r + c r • m ∆F i ) (16)
where:

• m ∆F i is the compressed size of the Bloom filter of the incremental dataset ∆F i (bits)

Since the size of the filters is small, these extra overheads are negligible compared to the overheads associated with redundant data in the incremental dataset.

Experimental evaluation for filter-based equijoins

In this section, we present experimental results obtained from the execution of two-way joins, chain joins, and recursive joins.

Two-way joins

Cluster environment and datasets.

All experiments were run on a cluster of 15 virtual machines using Virtualbox [START_REF]Oracle: Oracle vm virtualbox[END_REF]. Each machine has two 2.4Ghz AMD Opteron CPUs with 2MB cache, 10GB RAM and 100GB SATA disks. The operating system is 64-bit Ubuntu server 12.04, and the java version is 1.7.0.21. We installed Hadoop [START_REF]Apache: Hadoop[END_REF] version 1.0.4 on all nodes. One of the nodes was selected to act as Master and ran the NameNode and the JobTracker processes; the remaining nodes host the TaskTrackers in charge of data storage and data processing. Each TaskTracker node was configured to run up to two simultaneous map tasks and two reduce tasks. The HDFS block size was set to 128MB, size of read/write buffer was 128KB, heap-size for JVMs was set to 2048M, and the number of reduce tasks set to 28. All test datasets were produced by a data generation script of the Purdue MapReduce Benchmarks Suite [START_REF] Ahmad | Puma benchmarks and dataset downloads[END_REF], called "PUMA" which represents a broad range of MapReduce applications with high/low computing requirements and high/low shuffle volumes. The maximum number of columns in the datasets is 39 and string length in each column is set to 19 characters. The first column of Dataset1 is a foreign key that refers to the fifth column of Dataset2. We used three test sets Test 1, Test 2, and Test 3 with respective sizes 30GB, 70GB, and 110GB. Table 4 summarizes the dataset sizes used in our experiments. The ratios of the joined records are 0.054% (Test 1), 0.057%(Test 2), 0.063%(Test 3).

We executed our algorithm for the following join query.

SELECT * FROM dataset1(column0..column20) d1, dataset2(column0..column20) d2 WHERE d1.column0 = d2.column5

We particularly investigate four aspects: the number of intermediate tuples generated, the total execution time, the tasks timeline, and the scalability measured by varying the input size.

Evaluation of approaches.

In order to execute the filter-based algorithms efficiently, we specified the size of filters according to the cardinality of the join key values of datasets and chose the largest filter. There is a tradeoff between m and the probability of a false positive. Hence, the probability of a false positive f is approximated by:

f ≈ 1 -e -ρ•n/m ρ
For a given false positive probability f, the size of the Bloom filter m is proportional to the number of elements n in the filter as shown in Table 5. where ρ is the number of hash functions, and m/n is the number of bits allocated to each join key. We can determine optimized parameters for the filter (e.g. f, ρ and m) [START_REF] Broder | Survey: Network Applications of Bloom Filters: A Survey[END_REF]. In practice, however, we should choose values less than an optimized value to reduce computational overhead. As shown in Table 5, we deliberately select various values of f, ρ and m/n for the experiments to consider if they might affect our join performance. The filter files generated in the tests are compressed with gzip. The intermediate data size (Map output) is given in Table 6. The Reduceside join (without filter) is the most inefficient solution, although it runs as a single job. This is correlated to the large size of intermediate data. Note that the number of intermediate tuples generated in this case is almost equal to the number of Map input records, see Table 4 andTable 6. This slight difference is because a few tuples of Dataset2 have less than 6 columns, and so they have been eliminated.

Filter-based joins are more efficient in general. bj and ifbj include the preprocessing job and the filtering operation to improve the join performance.

The number of intermediate tuples produced by bj is considerably reduced with respect to rsj. However, in comparison to ifbj (see in Table 6), bj still produces much more intermediate data because the filtering operation is only executed on one input dataset (Dataset1 ). This situation is overcome by ifbj.

Looking at bj and ifbj, Table 6. points out that bj generates more intermediate data than ifbj. Namely, for the 110GB test, bj produces 145,206,430 intermediate tuples, whereas ifbj produces 179,091 tuples. The experiments reported above are consistent with our theoretical analysis (Theorem 1).

Next, we evaluate the efficiency of these join algorithms by comparing the total execution time. As a general fact, the join algorithms generating less intermediate data turn out to be faster, even if we sum up the cost of the pre-processing and join jobs. Table 7 gives the total execution time of the pre-processing job and the join job for each algorithm. Regarding pre-processing, the cost of the filter-based joins is related to the size of the data accessed to build the filter(s). In particular, ifbj has to scan two input datasets. However, it pays off, since once the filters are available, the cost of join jobs is drastically reduced.

Fig. 8 demonstrates that the best execution results from using intersection filters. Their total execution time is significantly reduced compared to bj in spite of the time spent in the pre-processing job. The total execution time of ifbj increases from about 10 to 105 (mns), whereas that of bj ranges from 19.19 to 145.20 (mns). The worst execution is rsj, ranging 28.25 to 150 (mns). The smaller cost of ifbj compared to the others (Table 7), is analyzed in Theorem 2. Finally, we analyze the sequence of tasks during job execution (called task timelines). We do not examine the task timelines of the pre-processing job which is negligible compared to the join query over large datasets (see Table 7).

Fig. 9 represents the task timelines of 70GB join jobs. These graphs are created by parsing log files generated by Hadoop during the job execution (555 map tasks and 28 reduce task, processing 185,205,828 input records and producing 26,062,967 output records). Each graph shows the respective timelines for map, shuffle and reduce phases.

There is a notable difference between the task timeline of ifbj and that of other joins. The execution time of all map and reduce tasks of ifbj, Fig. 9 (a), is significantly reduced compared to bj and rsj, Fig. 9 (b) and(c). Besides, the map and reduce phases of ifbj finished earlier than bj and rsj because they produce less intermediate data and, as a consequence, the total cost of the local I/O, sort, and remote data copy is also smaller. Joins that use the intersection filter are the most efficient solutions because of their better data filtering efficiency.

The efficiency of filter-based joins depends on the ratio of non-matching tuples. The threshold is defined by the two parameters δ dataset2 and δ dataset1 , which are the ratios of matching tuples. Fig. 10 shows the execution time of algorithms for several values of these parameters, in order to identify their impact. We start with an extreme case (first column) where domain of join attributes in Dataset1 and Dataset2 are disjoint. A ifbj is then able to discover the empty and the join job can therefore be omitted altogether, and their costs represents only that of the pre-processing job. This cost is roughly comparable with that of rsj because of the small size of the dataset which make the join job fast enough. Filtered joins should not be used for small input datasets because the cost of building and broadcasting filter(s) becomes relatively significant.

We next examine the cases of a high ratio of matching tuples (85% : 4%) and (95% : 65%). They represent respectively the thresholds for filter-based join resulting from our analysis. Fig. 10 clearly shows that this is the point where filters become counter-productive. This can be determined at compile time based on the ratios δ dataset2 : δ dataset1 .

The last case shows a join between fully matching datasets (100% : 100%), in which case rsj is the best solution.

Multi-way joins

Cluster environment and datasets.

We run experiments for the chain join on another computer cluster of 15 virtual machines using KVM (Kernel-based Virtual Machine) [START_REF]KVM: Kernel virtual machine[END_REF]. Each machine has two 1.4Ghz AMD Opteron CPUs with 512KB cache, 5GB RAM and 100GB SATA disks. We installed Hadoop [START_REF]Apache: Hadoop[END_REF] version 1.0.4 on all nodes. The other configurations of this cluster are similar to the ones of the cluster running the experiments of the two-way joins. The number of reduce tasks is set to 25.

All datasets were also produced by the data generation script of the PUMA. The maximum number of columns in the datasets is 39 and string length in each column is set 19 characters. The datasets Dataset1, Dataset2, Dataset3, and Dataset4 contain the join key columns column1 (c 1 ), column2 (c 2 ), column3 (c 3 ), and column4 (c 4 ). Tables 8 and9 summarizes the different input dataset sizes and the joined record ratios, resp. The chain join algorithms developed in our experiments are the Reduce-side join cascade (cj-rsj), the Bloom join cascade (cj-bj), the IF-based join cascade (cj-ifbj), the optimized two-way join cascade (ocj-2wj), and the optimized three-way join cascade (ocj-3wj). The following chain join query is used.

SELECT * FROM dataset1(c1..c10) d1, dataset2(c1..c10) d2, dataset3(c1..c10) d3, dataset4(c1..c10) d4 WHERE d1.c2 = d2.c2 AND d2.c3 = d3.c3 AND d3.4 = d4.c4

Evaluation.

The experiments use the parameters of the Boom filters given in Table 10. In order to confirm the cost model of chain joins (Section 4.2), we first examine the amount of intermediate data (Table 11) Next, we examine the total output of the chain join algorithms (Fig. 12). The total output consists of all the intermediate data generated in the map phase and the intermediate join results. In other words, it includes all map output tuples and reduce output tuples produced during the chain join. As shown in Fig. 12, cj-rsj and cj-bj generate the largest outputs; whilst the OCJ joins (e.g. ocj-2wj and ocj-3wj) using the extended intersection filters produce the least output. The cj-ifbj joins generally produce a little more output than the OCJ joins. The main reason is that the OCJ joins have the ability to filter out much more non-matching tuples than the others.

Both cj-rsj and cj-bj exhibit a similar pattern, with a significant cost increase from 26GB to 100GB. Obviously, cj-rsj has the highest cost with 119,928,957 tuples for Test 1, (77,035,830 for cj-bj and 32,942,272 for the cjifbj joins). This is even worse with Test 3, cj-rsj produces 371,782,345 tuples compared to 170,448,392 for cj-bj and 82,697,894 for the cj-ifbj joins.

Let us finally discuss the performance comparison, summarized by Fig. 13. The run time is clearly correlated to the size of the intermediate data, as confirmed by the comparison of the relative performance of the algorithms and the number of tuples shipped during the execution of joins. The two bottom graphs show the total execution times of the OCJ joins (ocj-3wj and ocj-2wj), the next three ones deal with cj-ifbj, and the two top graphs show cj-bj and cj-rsj. For the largest dataset (100GB), ocj-3wj and ocj-2wj run time is about 52.57 and 57.22 minutes respectively, while the cj-ifbj joins run time is about 65.13 minutes. cj-bj and cj-rsj run time is much longer, about 72.09 and 88.34 minutes resp. This shows the high benefit of filtering out useless data, as carrying this data all over the process constitutes a strong penalty. Note that these costs include the pre-processing step for filterbased joins. In a scenario where filters are built once, and the joins processed many times, the benefit of the approach is even reinforced. The results of these experiments are consistent with our cost analysis presented in Section 4.2.

Recursive joins

Cluster environment and datasets.

We performed experiments on a HaLoop cluster running the modified version of Hadoop 0.20.2. The cluster consists of 12 PC computers. Each machine has two 2.53GHz Intel(R) Core(TM)2 Duo CPUs with 3MB cache, 3GB RAM and 80GB SATA disks. The operating system is 64-bit Ubuntu server 14.04 LTS, and the java version is 1.8.0.20. This cluster has one TaskTracker and one DataNode daemon running on each node. One of the nodes is selected to act as a master and run the NameNode and the JobTracker processes. TaskTracker nodes are configured to run up to two simultaneous map tasks and two reduce tasks. The HDFS block size was set to 128MB, size of read/write buffer was 128KB, and the number of reduce tasks is set to 16.

We use test datasets generated by the PUMA to conduct the experiments. The maximum number of columns in the datasets is 31 and string length in each column is set 19 characters. The input dataset Know contains two join key columns, namely, column0 (c 0 ), and column1 (c 1 ). The following recursive join query is used to evaluate our experiments.

F riend(c 0 , c 1 , . . . , c 30 ) ←-Know(c 0 , c 1 , . . . , c 30 )

F riend(c 0 , c 1 , . . . , c 30 ) ←-F riend(c 0 , c 1 , . . . , c 30 ) c1=c 0 Know(c 0 , c 1 , . . . , c 30 )

Evaluation.

The filters' parameters used in the filter-based approach are listed in Table 13. We first examine the total map output (Table 14). The Shaw's approach (rejshaw) generates more intermediate data than the filter-based approach (rejfb). For the tests from 10GB to 30GB, rej-shaw generates from 215,609,705 to 602,707,978 tuples, whilst rej-fb has less than from 188,597,706 to 527,220,188 respectively. This is because the intermediate data of the join jobs in rej-shaw contains a lot of non-matching tuples, whereas rej-fb uses the intersection filter to eliminate these non-matching tuples from the intermediate data of the join jobs. Next, we examine the efficiency of the recursive join approaches. The total execution time of rej-shaw is compared to that of rej-fb. Let us look in Fig. 14 for more details. With the 10GB input dataset Know, the total execution time of the Shaw's approach is higher than that of rej-fb. This remains so through the other tests.

Conclusions and future work

The join operation is one of the essential operations for data analysis. Join evaluation is expensive and not straightforward to implement with MapReduce. This paper makes three contributions. First, we attempt to gather in a uniform setting some of the main approaches recently proposed for the most common types of joins. In particular, we systematically considered the introduction of filters in execution plans. Filters are known to greatly improve the cost of distributed joins thanks to their ability to avoid network transfer of useless data. We showed how to adapt the join algorithms with filters, on a systematic basis. The second contribution is a modeling of cost that serves as a yardstick to compare the expected efficiency of joins. In particular, we characterize the situations where filters are indeed beneficial. Finally, we conducted a full set of experiments to validate our models, and reported the behavior of the join algorithms in a practical situation.

In general, join evaluation using filters is more efficient than other solutions since it reduces the need for shipping non-matching data. Specific situations may lead to reconsider this general assumption. For instance, in the case of a join between two relations linked by an integrity constraint (primary, foreign key), the system guarantees the inclusion of one key set into the other, and filtering becomes useless. Such structured datasets are arguably not common in the Big Data realms. As another example, small dataset size may reveal the cost of producing and shipping the filters. A direct join approach should be used in that case (in fact using MapReduce for small datasets is probably not a good idea in the first place). Our cost models help to detect those special cases and adopt the proper evaluation strategy.

The present study could be extended in several directions. First, a complete coverage would include star joins, and in general joins amongst n relations linked by complex relationships. Given the complexity of matching such a general setting with a MapReduce framework, we consider that the set of joins cases investigated in what precedes constitute a satisfying set of primitives to start with. Regarding our experimental evaluation, we did our best to use the state-of-theart MapReduce framework (e.g., HaLoop). We note that some recent distributed engines (e.g., Spark [7], Stratosphere/Flink [START_REF]Stratosphere: Next generation big data analytics platform[END_REF] [START_REF]Apache: Flink[END_REF]) natively bring some of the features examined here, and notably iterations. At a physical level, they support caching of intermediate result, if possible in main memory. This strengthens our expectation that joins (including recursive joins) as studied here, constitute the basic building block of sophisticated algorithms for machine learning and data mining, which stand as the most promising outcome of Big Data processing in a near future. In this respect, the present study stands as a first step toward the design of an optimizer for distributed query processing, apt at considering complex integration of iterative, recursive and multi-set operators. We plan to investigate in the future the foundations of such an optimizer.

Fig. 1 .

 1 Fig. 1. MapReduce distributed execution

Fig. 2 .

 2 Fig. 2. Extended intersection filter -EIF (BF1, BF2, . . . , BFm)

Fig. 3 .

 3 Fig. 3. Distributing tuples of R, K, and L among r = n 2 reducers

Fig. 4 .

 4 Fig. 4. Three-way join extended with intersection filters

  R 1,2,••• ,i are filtered by the filters, whereas the base relations R i are not, where i ∈ [2, n].

Fig. 5 .

 5 Fig. 5. Implementation of a chain join using a cascade of two-way joins using intersection filters (cj-ifbj)

Fig. 6 .

 6 Fig. 6. Optimization of a chain join using extended intersection filters (OCJ)

Fig. 7 .

 7 Fig. 7. Semi-naive implementation of recursive joins in MapReduce

  ) holds for ifbj, denoted D ifbj • equation (3) holds for bj, denoted D bj • equation (4) holds for rsj, denoted D rsj • f I (R, L) is the false positive probability of the intersection filter IBF (R, L) [30], • and f (L) is the false positive probability of the Bloom filter BF (L).

Theorem 4 .

 4 A 3wj, R(A, B) K(B, C) L(C, D), is more efficient than a cascade of 2 two-way joins (R(A, B) K(B, C)) L(C, D) or R(A, B) (K(B, C) L(C, D)) when r < (|R|.α) 2 . Additionally, the size of the intermediate data is specified by|D| = 2 • |R| • √ r , for 3wj. |R| 2 • α, for a cascade of 2 two-way joins.where r is the number of reducers, |R|= |K|= |L|, and α is the probability of two tuples from different datasets to match on the join key column.Proof. Similar to the proof of Afrati and Ullman in[START_REF] Afrati | Optimizing joins in a map-reduce environment[END_REF]. First, we consider 3wj. Two attributes B and C are join key columns. We use hash functions to map values of B to b different buckets, and values of C to c buckets, as long as b • c = r. The intermediate data size of the three-way join is

  and |R| • c = λ • r We can multiply these two equations together to get |L| • |R| = λ 2 • r From here, we deduce λ = |R| • |L|/r By substituting the value of λ in the Lagrangian equations, we get: b = |R| • r/ |L|, and c = |L| • r/ |R| Then, from expression (11), we get the minimum communication cost of 3wj

Fig. 8 .

 8 Fig. 8. Total execution time

Fig. 9 .

 9 Fig. 9. 70GB Task timelines during the execution of the join job

Fig. 10 .

 10 Fig. 10. Identification of threshold for non-matching tuples for joins with 2GB inputs

Fig. 11 .

 11 Fig. 11. Total intermediate data of the chain join

  Fig. 11 helps us to obtain a visual comparison of the intersection filter-based chain joins. ocj-3wj has the least amount of intermediate data because it has the least number of jobs, and filters out almost all non-matching tuples in intermediate results. The intermediate data amount of ocj-2wj is slightly greater than the intermediate data amount of ocj-3wj, as analyzed by Theorem 5. However, ocj-2wj is still better than cj-ifbj chain joins which do not fully prevent non-matching tuples to propagate throughout the join chain.

Fig. 12 .

 12 Fig. 12. Total output data (Map output + Reduce output)

Fig. 13 .

 13 Fig. 13. Total execution time

Fig. 14 .

 14 Fig. 14. Total execution time

Table 1 .

 1 List of abbreviations

	Abbreviation Algorithm
	ifbj	Intersection filter-based join
	bj	Bloom join
	rsj	Reduce-side join
	3wj

Table 2 .

 2 Parameters of the cost model for two-way joins

	Parameter Explanation
	|R|	The size of R
	|L|	The size of L
	|D|	The size of the intermediate data
	c l

Table 3 .

 3 Parameters of our cost model for recursive joins

	Parameter Explanation
	c l

Table 4 .

 4 Input datasets

	Inputs	size	Test 1 records	size	Test 2 records	size	Test 3 records
	Dataset1	15GB 40,259,163	35GB	92,681,333	55GB 145,099,559
	Dataset2	15GB 40,108,215	35GB	92,524,495	55GB 139,573,823
	Total	30GB 80,367,378	70GB 185,205,828	110GB 284,673,382

Table 5 .

 5 Parameters for filters

	Tests	f	ρ	n	m/n	m (bit)
	Test 1	0.001	7	14,866	15	222,990
	Test 2	0.0001	8	15,790	21	331,590
	Test 3	0.0001	8	15,790	21	331,590

Table 6 .

 6 Number of intermediate tuples (Map output)

	Join algorithms	Test 1 (30GB)	Test 2 (70GB)	Test 3 (110GB)
	ifbj	43,453	106,116	179,091
	bj	40,276,915	92,747,151	145,206,430
	rsj	80,320,684	185,098,062	284,510,488

Table 7 .

 7 Execution of pre-processing job and join job (in minutes)

	Joins	Test 1 (30GB) Pre-Join Total	Test 2 (70GB) Pre-Join Total	Test 3 (110GB) Pre-Join Total
		proc.	job	time	proc.	job	time	proc.	job	time
	ifbj	3.17	6.15	9.32	6.45 24.25 31.10	10.00	92.12	102.12
	bj	2.12 17.07 19.19	3.63 43.63 47.26	5.22 139.58	145.20
	rsj	0 28.25 28.25	0 70.13 70.13	0 150.00	150.00

Table 8 .

 8 Input datasets used in three tests

	Inputs	size	Test 1 records	size	Test 2 records	size	Test 3 records
	dataset1	10GB 26,836,497	20GB	53,675,946	20GB	53,682,929
	dataset2	3GB	8,051,454	10GB	26,838,960	30GB	73,881,305
	dataset3	10GB 26,836,497	20GB	53,675,946	20GB	53,682,929
	dataset4	3GB	8,051,454	10GB	26,838,960	30GB	73,881,305
	Total	26GB 69,775,902	60GB 161,029,812	100GB 255,128,468

Table 9 .

 9 The ratios of the joined records of the datasets (%)

	Inputs	Test 1	Test 2	Test 3
	dataset1	0.721722639	0.304090688	0.123521020
	dataset2	0.216530370	0.152050936	0.169996205
	dataset3	0.721722639	0.304090688	0.123521020
	dataset4	0.216530370	0.152050936	0.169996205

Table 10 .

 10 Parameters of Bloom filters

	Tests	f	ρ	n	m/n	m (bit)
	Test 1	0.000101	8	13,147	21	276,087
	Test 2	0.000101	8	13,840	21	290,640
	Test 3	0.000101	8	15,295	21	321,195

Table 11 .

 11 Number of intermediate tuples (all map outputs)

	Chain join	Test 1	Test 2	Test 3
	algorithms	(26GB)	(60GB)	(100GB)
	cj-ifbj	1,309,349	1,469,048	1,497,692
	cj-bj	45,402,907	89,201,979	89,248,190
	cj-rsj	88,296,034	196,465,292	290,582,143
	ocj-2wj	1,281,036	1,417,684	1,445,428
	ocj-3wj	1,221,769	1,359,575	1,385,053

Table 11

 11 shows that cj-rsj and cj-bj generate much more intermediate data than any algorithms using the (extended) intersection filters.

  Table 12 lists the different sizes of the dataset Know used in our tests.

Table 12 .

 12 Input dataset Know with different sizes

	Test	Size	Records
	Test 1	10GB	53,674,078
	Test 2	20GB	107,349,426
	Test 3	30GB	150,000,054

Table 13 .

 13 Parameters of filters

	Tests	f	ρ	n	m/n	m (bit)
	Test 1	0.000101	8	7,111	21	149,331
	Test 2	0.000101	8	7,123	21	149,583
	Test 3	0.000101	8	7,130	21	149,730

Table 14 .

 14 The total number of intermediate tuples

	Recursive join	Test 1	Test 2	Test 3
	approaches	(10GB)	(20GB)	(30GB)
	rej-shaw	215,609,705	431,589,879	602,707,978
	rej-fb	188,597,706	377,403,437	527,220,188

Our study only considers conditions is based an equality operator (=), or equijoins.

Thuong-Cang Phan, Laurent d'Orazio, Philippe Rigaux