The Web of Data contains a large number of di erent, openlyavailable datasets. In order to e ectively integrate them into existing applications, meta information on statistical and structural properties is needed. Examples include information about cardinalities, value patterns, or co-occurring properties. For Linked Datasets such information is currently very limited or not available at all. Data pro ling techniques are needed to compute respective statistics and meta information. However, current state of the art approaches can either not be applied to Linked Data, or exhibit considerable performance problems. We present Lodop, a framework for computing, optimizing, and benchmarking data pro ling techniques based on MapReduce with Apache Pig. We implemented 15 of the most important data pro ling tasks, optimized their simultaneous execution, and evaluate them with four typical datasets from the Web of Data. Our optimizations focus on reducing the amount of MapReduce jobs and minimizing the communication overhead between multiple jobs. Our evaluation shows the signi cant potential in optimizing the runtime costs for Linked Data pro ling.
Over the past years, an increasingly large number of datasets has been published as part of the Web of Data. This trend, together with the inherent heterogeneity of datasets and their schemata, makes it increasingly time-consuming to nd and understand datasets that are relevant for integration. In order for users to be able to integrate Linked Data, they rst need an easy way to discover and understand relevant datasets on the Web of Data.
Data pro ling is an umbrella term for methods that compute meta-data for
describing datasets [
Use cases for data pro ling can be found in various areas concerned with
data processing and data management [
Query optimization is concerned with nding optimal execution plans for database queries. Cardinalities and value histograms can help to estimate the costs of such execution plans. Such metadata can also be used in the area of Linked Data, e.g., for optimizing SPARQL queries.
Data cleansing can bene t from discovered value patterns. Violations of detected patterns can reveal data errors, and respective statistics help measure and monitor the quality of a dataset. For Linked Data, data pro ling techniques help validate datasets against vocabularies and schema properties, such as value range restrictions.
Data integration is often hindered by the lack of information on new datasets. Data pro ling metrics reveal information on, e.g., size, schema, semantics, and dependencies of unknown datasets. This is a highly relevant use case for Linked Data, because for many openly available datasets only little information is available1.
Schema induction: Raw data, e.g., data gathered during scienti c experiments, often does not have a known schema at rst; data pro ling techniques need to determine adequate schemata, which are required before data can be inserted into a traditional DBMS. For the eld of Linked Data, this applies when working with datasets that have no dereferencable vocabulary. Data pro ling can help induce a schema from the data, which then can be used to nd a matching existing vocabulary or create a new one.
The process of running data pro ling tasks for large Linked Datasets can take hours to days, depending on the complexity of task and the size of the respective datasets. Data set characteristics highly in uence the pro ling task runtime. As an example, our Property Cooccurrence by Resource script (see Sec. 3) runs 16 hours for only 1 million triples of the Web Data Commons RDFa dataset in contrast to 5 min on Freebase and 9 min on DBpedia.
We have compiled a list of 56 data pro ling tasks implemented in Apache
Pig to be executed on Apache Hadoop. At this point Apache Pig only applies
some basic logical optimization rules, like removing unused statements [
While many tools and algorithms already exist for data pro ling in general, most of them can, unfortunately, not be used for graph datasets, because they assume a relational data structure, a de ned schema, or simply cannot deal with very large datasets. Nonetheless, some Linked Data pro ling tools already exist. Most of them focus on solving speci c use cases instead of data pro ling in general.
One relevant use case is schema induction, because not having a xed and
well-de ned schema is a common problem with Linked Datasets. One example
for this eld of research is the ExpLOD tool [
An example for the query optimization use case is presented in [
In [
This section lists and explains a set of useful data pro ling tasks to pro le Linked Data sets. We have implemented a total of 56 data pro ling scripts, which compute 15 di erent statistical properties across di erent subsets of the input dataset. These subsets are determined via the following types of groupings:
Resource: grouping based on the triple subject; usually paired with some top-k list of resources.
Class: grouping based on the value of the rdfs:type predicate on a resource. Class & Property: grouping based on both class and property type. Datatype: grouping based on the (declared) data type of the object value; only typed literals are considered.
Language: grouping based on the (declared) language of the object value; only language literals are considered.
Context URL: grouping based on the N-Quads context attribute. Values being aggregated based on the context URL are grouped three times: based on the full URL, the pay-level-domain (PLD) part of the URL, and the top-level-domain (TLD) part of the UR.
Vocabulary: grouping based on the vocabularies used for predicates or classes. Object URI: grouping based on the value of the object. Only URI values are considered.
The following data pro ling tasks are computed. Unless stated di erently, examples have been generated from a 1 million triples subset of DBpedia. More information on datasets can be found in Section 6.
Number of triples for each of the following groupings: overall, resource, property, object URI, context URL. This highlights, e.g., that the largest resources in our DBpedia subset is http://dbpedia.org/resource/2010-11_SC_Freiburg_ season with 867 triples, and the most used property type is rdfs:type with 86,856 triples. (7 scripts) Average number of triples per resource within each of the following groupings: overall, context URL, class. This reveals that resources in our DBpedia subset are made up of 5.9 triples on average, and that some classes, such as http://dbpedia.org/ontology/YearInSpaceflight, have resources with an average number of 502 triples. (5 scripts) Average number of triples per object URI highlights how often certain URI objects are used across the dataset. For our DBpedia subset, each URI object is (re-)used as the object value for an average of 2.6 triples. For our 1 million triples subset of Freebase, this number is lower (1.9), i.e., fewer URI object values are reused. (1 script) Average number of triples per URL computes the average number of triples per value of the graph context URL. For our DBpedia subset, each graph context contains an average of 47.8 triples. This means that, in this case, multiple resources share the same graph context, because the average number of triples per resource is only 5.9 as mentioned above. (1 script) Number of property types within each of the following groupings: overall, context URL, resource, class, data type. This tasks tells us how many property types are used in di erent contexts; our DBpedia subset has a total of 7,844 di erent property types, of which 4,970 are used on resources of the class owl: Thing. It also reveals that 2,929 property types point to triples having typed object values with the xsd:int datatype. (7 scripts) Average number of property values per property type (i.e., triples per predicate) within each of the following groupings: overall, context-URL, resource, class, property, class, property. For example, resources having the class http:// umbel.org/umbel/rc/Event have an average number of 115 values for the http: //dbpedia.org/property/time predicate in our DBpedia subset. (8 scripts) Number of resources in each of the following groupings: property, class, datatype, language, vocabulary. On our subset of DBpedia, these pro ling tasks tell us that there are a total of 169,035 resources, that the owl:sameAs property type is used on 43,840 resources, and the http://xmlns.com/foaf/0.1/Person class on 4,140 resources. (6 scripts) Number of context URLs in each of the following groupings: property, class, vocabulary. For DBpedia the context URL usually matches to the corresponding Wikipedia page. This task can reveal how many Wikipedia pages lead to certain classes; for example, the foaf:Person class occurs in 4,140 di erent context URLs in our subset of data. (3 scripts) Number of context PLDs in each of the following groupings: property, class, vocabulary. Alternative script versions additionally group by the TLD of the context URL. Similar to the previous task, this task can tell us how many di erent pay-level domains are responsible for certain properties, classes, or vocabularies. For our WDC RDFa subset (1m triples) this reveals, e.g., that the http://www.facebook.com/2008/ predicate is used by 452 domains, which makes it the most-used property type in this subset. (6 scripts) Property co-occurrence computes a list of pairs of property types that are used together: a) on the same resource (by resource), b) within the same context URL (by url), or c) pointing to the same resource (by object URI). This reveals patterns, such as http://dbpedia.org/ontology/artist and http: //dbpedia.org/property/title being used together on resources describing artistic work. (3 scripts) Inverse properties computes a list of properties that are inverse to each other. For example, the relationship between a musician and his band can be declared on both resources with two inverse property types http://dbpedia.org/ ontology/associatedBand and http://dbpedia.org/ontology/bandMember. (1 script) URI-literal ratio computes the ratio between the number of URI values and literals within the following groupings: overall, class, context URL. Our DBpedia subset contains almost as many URI object values as it contains literal values (ratio 1.1); however, the RDFa subset contains far more literals than URI values (ratio 0.2). (5 scripts) Property value ranges of literal values per property type (numeric and temporal values only). For example, the http://dbpedia.org/ontology/bedCount property has numeric values between 76 and 785 in our DBpedia subset. (2 scripts) Average value length of literals per property type can reveal schema properties. For example, the http://dbpedia.org/ontology/title property has values with an average length of 34 characters, whereas values for the http: //dbpedia.org/property/longTitle property have an average of 288 characters. (1 script) Number of inlinks computes the number of URI values pointing to other resources within the same dataset/ le. For DBpedia, there are 207,712 values pointing to resources within the same dataset; for our WDC RDFa subset, there are far fewer inlinks (35,329). (1 script) Each of these pro ling tasks have been implemented as an Apache Pig script and are availabe at https://github.com/bforchhammer/lodop. The runtime of these scripts even on 1 million triples might take up to hours, e.g., for the property co-occurrence determination. Also, scripts often have the same preprocessing steps, e.g., ltering or grouping the dataset. Thus there is a large incentive and potential to optimize the execution of multiple scripts. 4
A prevalent goal for relational database optimization is to reduce the amount of
required full table scans, which for le-based database systems e ectively means
reducing the amount of disk operations. Sellis introduces Multi-Query
Optimization for relational databases as the process of optimizing a set of queries which
may share common data [
Apache Pig3 is a platform for performing large-scale data transformations on top of Apache Hadoop clusters. It provides a high-level language (called Pig Latin) for specifying data transformations, e.g., selections, projections, joins, aggregations and sorting on datasets. Pig Latin scripts are compiled into a series of MapReduce tasks and executed on a cluster.
The main goals for our multi-query optimization rules for Pig are the following two: First, we attempt to minimize the data ow between operators. In our evaluation (Sec. 6) we identi ed the data ow between MapReduce jobs as a reasonable indicator for the performance of Pig scripts, as it is closely related to the amount of required HDFS operations. Second, we try to avoid performing identical or similar operations multiple times. The idea behind this is to free up cluster resources for other tasks. All optimization rules presented in this section, are based on optimizing the logical plans of Pig scripts.
Three optimization rules have been implemented: Rule 1 merges identical operators in logical plans of di erent scripts, Rule 2 combines FILTER operators, and Rule 3 combines aggregations, i.e., FOREACH operators. Rule 1 is a prerequisite for the other two rules, which work on pairs of siblings operators, i.e.,
operators that have the same parent operator in a respective logical plan. For all optimization rules, it was important to make sure that their usage does not a ect the intended output of scripts.
Rule 1 { Merge identical operators: In order to better utilize cluster resources, it makes sense to submit jobs to Apache Hadoop in parallel. Lodop supports this by merging logical plans of di erent scripts into a single large plan. In our experiments, executing scripts in parallel as part of one large plan cuts execution time down to 25-30% of the time required to execute scripts sequentially. Once all plans have been merged together, it's possible to also merge identical operators.For 52 of our Pig scripts, this reduces the number of operators from 365 to 267.
Rule 2 { Combine lters: FILTER operators reduce the amount of data that needs to be processed in later steps of the execution pipeline. This optimization rule aims to avoid iterating over large sets multiple times. From our selection of pro ling scripts, 25 scripts perform ltering operations on the full initial dataset.
First, we identify all suitable sibling lters, i.e., all FILTER operators that have the same parent operator. Second, a combined lter is created and we attach it to the same parent operator. This combined lter contains all boolean expressions of existing lters concatenated via OR. The expression of the combined lter is cleaned up by transforming it into disjunctive normal form. Finally, we rearrange all previous lters and move them after the combined lter. Rule 3 { Combine aggregations: FOREACH operators can be used for projections and aggregations. Some instances perform identical aggregations, but project di erent properties. This can happen, e.g., if the aggregation itself is only a preprocessing step to another aggregation. These operators are not exactly identical, so the rule for merging identical operators will not be able to merge them. However, these cases can be optimized by separating the aggregation from the projection, i.e., performing the aggregation only once with all projected columns, and then projecting the exact columns afterwards. For our set of scripts, this rule can be applied in seven di erent cases and combines varying numbers of FOREACH operators from the minimum of two to a maximum of eleven siblings operators.
While our goal is to optimize the performance of pro ling tasks, the optimization rules can be applied on any Pig script. 5
Lodop comprises four major components. Figure 1 shows how they interact with the Hadoop cluster. The Compiler is responsible for compiling Pig Latin scripts into Logical Plans and eventually into MapReduce jobs; the Optimiser takes care of optimizing logical plans; the ScriptRunner schedules and monitors the execution of jobs on the Hadoop cluster; nally, the Reporting component turns raw statistics into human-readable formats.
Lodop is built to be easily con gurable via command line arguments, allowing the execution of di erent combinations of datasets, scripts, and optimization rules. The benchmarking system parses the arguments supplied by the user, and passes them on to the compiler. The main responsibility of the compiler is to make sure that Apache Pig plays well with the optimization component. It takes over Pig's standard script compilation work ow, and adds in missing features, such as the support for multiple scripts and custom optimization rules.
The compiler rst loads the declared list of scripts and compiles them into respective logical plans. This compilation step is largely taken care of by Apache Pig's compiler, which parses Pig Latin and builds a respective directed acyclic graph of logical operations. Once these logical plans have been created, our standardized loading and storage functions are injected into each plan. The loading function assumes that all scripts work with the same input le type and schema (N-Quads les). This is an essential prerequisite for merging di erent scripts together in the optimization component. Similarly, outputs are also handled automatically, by always storing the result of the last statement in each script. The respective storing function writes results directly to Hadoop's distributed le system (HDFS).
Rule-based optimization is now done in two steps: First, all existing logical plans are merged into one large logical plan. At this point, each script's logical data ow is still separate from other scripts, but this allows the ScriptRunner component to submit jobs for di erent scripts at the same time, and thus execute them in parallel. Second, optimization rules are applied to the merged logical plan. The rules function similar to other optimization rules that are already part of Pig: First the logical plan is searched for applicable patterns in order to gain a list of possible optimization targets. Each match is then checked to determine whether the rule can be applied to the speci c group of operators. Finally, if all checks are positive, the rule is applied and the logical plan adjusted accordingly.
Lodop currently does not perform cost-based optimization to determine whether a rule should be applied or not; we simply apply rules repeatedly until they cannot be applied any more. To ease debugging and help understand logical plans, the system additionally visualizes plans after di erent steps in the compilation and optimization process.
After the optimization step, logical plans are compiled into MapReduce plans and then handed to the ScriptRunner component. The ScriptRunner submits MapReduce jobs to the Hadoop cluster, monitors their execution, and gathers performance statistics. Scheduling and monitoring are handled by Apache Pig itself. Statistics on the performance of MapReduce jobs are provided by Hadoop and only need to be retrieved from the cluster. 6
In this section we evaluate the e ect of applying the optimization rules of Sec. 4 and investigate under which circumstances the performance is improved. We evaluate on four selected Linked Data sets that provide a wide range of characteristics, ranging from cross-domain to domain-speci c as well as ranging from well-de ned to loosely de ned semantics. 6.1
Datasets and experimental setup DBpedia4 is one of the largest Linked Data sets and contains structured information extracted from Wikipedia. As such it covers a wide range of di erent topics with a large number of property types and a very large number of classes compared to other datasets. Almost 50% of property values are URIs.
Freebase5 is a community-maintained database of \well-known people, places, and things". It contains data harvested from sources such as Wikipedia, ChefMoz, MusicBrainz, and others. It has a fairly small average number of triples per resource (4.4), but the highest number of URI values.
The Web Data Commons6 project extracts RDF triples directly from information embedded on websites via RDFa, Microdata, or Microformats. We focus on only the RDFa subset (roughly 500 million triples) which is mostly unstructured and has a small schema compared to other datasets. Most property values are literal values (83%) and there are very few in-links (3%). Any faulty de nitions on crawled websites can lead to inconsistencies and errors in the triple graph, e.g., resources that do not conform to well-known schema de nitions.
The species dataset of the European Environment Agency (EUNIS)7 is part of a database on species, habitat types, and sites of interest for biodiversity.
The species subset is well-structured with only one class and 16 property types. Compared to other datasets it has a very large average number of triples per resource (15.2).
All evaluations were performed on an Apache Hadoop cluster with one head node and ten slaves. Each node has a 2-core processor (Core 2 Duo) with 2GB of memory and runs CentOS 5.5. We use Apache Hadoop 1.1.2 and Apache Pig 0.11.1 with Java 1.7. In order to explain our observations, we particularly look at the properties identi ed as relevant to performance, i.e., the number of operators and MapReduce jobs, as well as the amount of I/O activity on HDFS. 6.2
Base performance analysis This section analyzes the overall runtime of scripts, when executed without any custom optimizations. We thus establish a baseline against which our custom optimization rules can be compared, and give insights into the reasons for current performance.
Figure 2 shows an overview of execution times for all scripts executed on 1 million triples of DBpedia, Freebase, WDC RDFa, and EUNIS. The gure shows that execution times vary depending on the dataset and the script. Overall, one can see that many scripts nish in under 5 minutes. Some scripts have to perform more complex computations and hence take longer to complete. This includes the scripts for computing property co-occurrence and the URI-Literal ratio by Class script. These UDF-based scripts dominate the overall execution time, are not amenable to our rules, and are thus excluded from further evaluation.
E ectiveness of merging identical operators As described in Section 5, Lodop supports two ways for executing multiple scripts: First, they can simply be loaded, compiled, and executed sequentially. In this case, the monitoring and scheduling overhead of Apache Pig takes some additional time between jobs and also between scripts, during which the cluster is idle. In order to better utilize cluster resources, it makes sense to submit jobs to Apache Hadoop in parallel instead. Lodop supports this by merging logical plans of di erent scripts together into one large plan. By executing multiple jobs simultaneously, Apache Pig can start a MapReduce job, as soon as all its dependent jobs have nished. Figure 3 shows that this has a large impact on the overall runtime of the pro ling process, cutting execution time down to 25-30% of the execution time required to execute scripts sequentially.
Once all plans have been merged, it is possible to apply the rst optimization rule, merge identical operators. For 52 scripts, this reduces the number of operators from 365 to 267. Table 1 shows that we are able to merge, amongst others, 14 COGROUPs, 12 FOREACHs and 1 JOIN into respective identical operators. In terms of MapReduce jobs, merging identical operators reduces the number of jobs from 176 jobs to 140.
Unfortunately, this operator merging has little to no e ect on the overall performance, as can be seen in Figure 3. The following observations give reasons for this: First, we end up with a tree of MapReduce jobs after merging identical operators; while this was an intended e ect of this optimization rule, it also means that there is only one root node, which all other jobs depend upon. Because Apache Pig tries to execute as many operators as possible on early Hadoop jobs, we end up with one very large job at the beginning of the work ow. The reason for this is the process used by Apache Pig to translate logical plans into MapReduce plans: it generates one MapReduce job for each COGROUP operation, then moves most other operators into the reduce function of the previous job; all operators before the rst COGROUP are pushed into the map function of the rst job. Executing this large initial job can take up a signi cant amount of time. For example, for 1 million triples of the DBpedia subset, executing the merged root node takes almost 10 minutes (570s), which is about 30% of the total execution time for the merged plan. Compared to the simultaneous execution of unmerged scripts, this hinders parallelism.
Note that the large initial job is also the reason for the missing execution time value on the EUNIS species dataset in Figure 3: the respective cluster node ran out of memory during the computation of the respective rst job.
Second, the amount of data being transferred between jobs can actually increase. When scripts are executed simultaneously in the merged plan, multiple jobs load and process the same input le. Most scripts in this case manage to reduce the input size signi cantly during their respective rst job already. In contrast, when all identical operators are merged, only the root node loads the full input le. However, in this case 16 of the 36 children of the root job still need to work with the full number of input tuples as well. Most of them will have some columns projected out, so in terms of transferred bytes it is not the full dataset, but compared to executing scripts in parallel more data needs to be materialized on HDFS.
Merging identical operators is a prerequisite step to applying the remaining two optimization rules. However, it may be more bene cial to only combine identical operations based on a cost-based approach, which decides whether it is worth merging identical operators based on the possibility of applying other rules and based on an estimated performance gain. 6.4
E ectiveness of combining lters In order to evaluate the e ect of combining FILTER operators, we look at an example of two scripts: Property value ranges (temporal) and Property value ranges (numeric). Both scripts load the full input dataset and lter out only triples with typed object values. The former script further restricts the set by temporal data types, the latter by numeric data types. The two can be optimized by ltering the input dataset only once and applying the additional restriction on the data type afterwards.
Most of the datasets evaluated have only a small set of triples with typed object values (< 15%). Therefore, ltering the input dataset only once signi cantly reduces the selectivity for the additional selections. Statistics on the amount of HDFS I/O for this example show that the un-optimized plan reads almost twice as much data from HDFS than the optimized version (see Table 2). Yet Figure 4 shows that the e ect on the overall execution time is small.
Even though the merging of lter operators does not improve on the amount of HDFS I/O compared to merging identical operators, the graph still shows a consistent improvement of about 10-20s, compared to only merging identical operators. This can be explained by improved CPU time, i.e., less time spent on actual computation on the respective cluster node. In fact, the total amount of seconds spent on computations (additive, not considering simultaneous execution) is reduced noticeably: For 10M triples of the DBpedia subset, the CPU time is reduced from 390s for the merged plan to about 240s for the plan with merged identical operators and nally to about 220s for the plan with combined lter statements. While this presents an improvement compared to the plan with identical operators merged, it does not show a signi cant improvement on the runtimes of the merged plan, which executes scripts in parallel.
E ectiveness of combining aggregations In order to evaluate the e ect of combing FOREACH operators we regard three closely related example scripts: Average number of property values by class, by property, and by class and property. All three group the input dataset by subject and predicate and then count the number of triples in each group. The only di erence in this aggregation step is the projected columns for each of the scripts.
The goal of combing aggregations is to reduce the amount of data required to be read from HDFS, by having only one combined FOREACH operator iterating over the full input dataset. Table 3 shows that this is achieved for the given example. Compared to the merged plan, the amount of data being read from as well as being written to HDFS is reduced considerably for all 10M datasets. As explained earlier, execution plans for which we only merged identical operators exhibit higher amounts of I/O activity. Figure 5 shows that this optimization rule improves overall runtime for all datasets by 1-2 minutes.
It should be noted that the amount of HDFS activity is improved far less for other combinations of scripts. So, while this example demonstrates the potential bene t of combining aggregations, it also motivates the need for an advanced, cost-based optimizer that applies optimizations only in certain cases. Dataset Merged Plan Identical operators Aggregations combined merged DBpedia 10M 19.9GB / 14.6GB 22.4GB / 14.0GB 15.6GB / 11.8GB Freebase 10M 11.7GB / 7.8GB 14.2GB / 7.0GB 8.8GB / 5.2GB WDC-RDFa 10M 12.4GB / 7.5GB 15.4GB / 6.8GB 8.8GB / 4.6GB Table 3. HDFS I/O statistics for Average number of property values by class / property / class and property scripts. The number of MapReduce jobs and the amount of data ow in the operator pipeline are good indicators for the performance of Apache Pig scripts. Our evaluation shows that improving only on these factors does not necessarily improve overall performance. Merging identical operators reduces both the total number of operators and the number of MapReduce jobs. It comes at the cost of less parallelism. Combining lter operators was shown to reduce the execution time of map/reduce functions (i.e., CPU time). Combining aggregations can reduce the amount of HDFS I/O, and improves overall execution time for certain combinations of scripts and datasets. Figure 6 shows execution times when optimizations are applied for all scripts. Overall in our experiments, executing scripts in parallel and applying all optimization rules cuts execution time down to 25-30% of the time required to execute scripts sequentially.
Existing work on data pro ling can either not be applied to the area of Linked Data or exhibits considerable performance problems. To overcome this gap we introduced a list of 56 data pro ling tasks, implemented them as Apache Pig scripts, and analyzed the performance of executing them on an Apache Hadoop cluster. Thereby, we introduced three common techniques for improving performance, namely algorithmic optimization, parallelization and multi-script optimization. We experimentally demonstrated that they achieve their respective goals of optimizing the amount of MapReduce jobs or the amount of data materialized between jobs.
In addition to the optimization techniques described in this paper, other optimization rules could be implemented. For instance, simple projections could be ignored in the logical optimization plans to allow further merging of operators, and aggregations based on similar groupings could be optimised to reduce the amount of redundant computation. Further, an advanced optimization strategy should apply optimization rules only if the estimated overall performance gain is high enough and not as often as possible, as we do it now.