PaperlayerView exact revision on arXiv

Source-linked AI summary

Load Balancing for MapReduce-based Entity Resolution

Lars Kolb, Andreas Thor, Erhard Rahm

arXiv:1108.1631v1cs.DC

TL;DR

The paper addresses load imbalance in MapReduce-based entity resolution caused by skewed blocking blocks and the resulting concentration of matching work. It introduces a preprocessing-guided workflow with BlockSplit and PairRange, then evaluates both approaches on real-world data in a real cloud environment. Both approaches handle skew and distribute workload across reduce tasks, while PairRange is less dependent on initial input partitioning and slightly more scalable for large match tasks.

  • Problem

    Skewed entity-resolution blocks can assign excessive matching work and memory demands to individual reduce tasks, undermining MapReduce scalability and increasing cloud costs.

  • Method

    The workflow builds a block distribution matrix in a preprocessing MapReduce job and uses BlockSplit or PairRange to distribute large-block work across reduce tasks.

  • Results

    Both approaches are robust against data skew and scale with available nodes; PairRange is less dependent on initial input partitioning and slightly more scalable for large match tasks.

  • Takeaways & Limitations

    Load-balanced blocking enables parallel entity resolution despite skewed block distributions, with BlockSplit offering a simpler approach and PairRange stronger scalability for large match tasks.

Abstract

from arXiv · show

The effectiveness and scalability of MapReduce-based implementations of complex data-intensive tasks depend on an even redistribution of data between map and reduce tasks. In the presence of skewed data, sophisticated redistribution approaches thus become necessary to achieve load balancing among all reduce tasks to be executed in parallel. For the complex problem of entity resolution, we propose and evaluate two approaches for such skew handling and load balancing. The approaches support blocking techniques to reduce the search space of entity resolution, utilize a preprocessing MapReduce job to analyze the data distribution, and distribute the entities of large blocks among multiple reduce tasks. The evaluation on a real cloud infrastructure shows the value and effectiveness of the proposed load balancing approaches.

I. INTRODUCTION

MapReduce-based entity resolution faces severe load imbalance when skewed blocking blocks assign excessive work to individual reduce tasks. The paper introduces two skew-aware strategies that redistribute large blocks and evaluates them on real cloud data.

  • Motivation: Entity resolution identifies records referring to the same real-world object, but naïve comparison of n entities has quadratic complexity O(n^2).Blocking reduces the search space by partitioning entities according to one or more blocking-key attributes.
  • Motivation: Basic MapReduce entity resolution assigns each block to one reduce task, so skewed block sizes can create runtime, scalability, memory, and cloud-cost problems.A large block may prevent effective use of other nodes while its entities must be compared with one another.
  • Approach: The proposed workflow uses a preprocessing MapReduce job to build a block distribution matrix describing entities per block across input partitions.The matrix guides fine-tuned redistribution for subsequent parallel block matching.
  • Approach: BlockSplit assigns whole blocks when feasible and splits larger blocks into smaller chunks across input partitions for parallel matching.Its assignments respect load-balancing and memory constraints.
  • Approach: PairRange enumerates entity pairs and redistributes entities so reduce tasks perform approximately equal numbers of comparisons.The evaluation compares both approaches with each other and with the basic MapReduce strategy using real-world data in a real cloud environment.

III. LOAD BALANCING FOR ER

The paper presents a two-job MapReduce workflow that analyzes block distributions and redistributes entities from large blocks across reduce tasks to address skew and memory problems.

  • General ER Workflow: The workflow assumes one entity source R with valid blocking keys and produces matching entity pairs as output.
  • General ER Workflow: Basic assigns entities sharing a blocking key to one reducer, but skewed block sizes can dominate execution time and cause memory problems.A reducer must store all entities in its block in main memory or use slower external memory.
  • General ER Workflow: The proposed workflow uses two MR jobs: the first computes a block distribution matrix, and the second uses it to tailor entity redistribution.Both jobs use the same map-task count and input partitioning.
  • Load-Balanced Matching: The second job’s map phase creates composite keys encoding target reducers, blocks, and entities so associated partitioning and grouping functions can balance load.The workflow also carries annotated original entities into the second MR job.
  • Block Distribution Matrix: The block distribution matrix records entity counts for each block across input partitions and is computed by grouping composite keys containing blocking keys and partition indices.The first job outputs triples containing the blocking key, partition index, and entity count.

IV. BLOCK-BASED LOAD BALANCING

BlockSplit balances block-based entity-resolution work by splitting large blocks into sub-block match tasks, while PairRange balances work by assigning nearly equal ranges of enumerated pairs to reduce tasks.

  • BlockSplit: BlockSplit splits large blocks into m sub-blocks and creates match tasks for individual sub-blocks and their Cartesian products.Small blocks remain single match tasks; split-block entities are replicated to support the resulting tasks.
  • BlockSplit: BlockSplit orders match tasks by comparison count and greedily assigns the largest tasks first to reduce tasks with the least assigned work.Its composite keys encode the target reduce task, block, and split information while preserving block grouping.
  • BlockSplit: In the running example, splitting block Φ3 into two sub-blocks produces three match tasks requiring 1, 6, and 3 comparisons, distributed across three reduce tasks.Replicating the five entities in the split block increases the output to 19 key-value pairs from 14 input entities.
  • PairRange: PairRange virtually enumerates entities and comparisons, then assigns nearly equal contiguous ranges of all pair indices to the reduce tasks.The method uses the BDM to compute global entity indices and identify which ranges contain pairs involving each entity.
  • PairRange: For the example’s 20 pairs and three reduce tasks, PairRange assigns ranges [0, 6], [7, 13], and [14, 19].Entities are sent to every reduce task whose assigned range contains at least one relevant pair.

VI. EVALUATION

The evaluation varies data skew, configured map and reduce tasks, and available cloud nodes using two real-world datasets on Amazon EC2.

  • Evaluation design: The evaluation examines data skew, the numbers of map and reduce tasks, and the number of available cloud nodes while holding the other factors constant.These experiments assess the strategies across three performance-critical dimensions.
  • Evaluation environment: Experiments run on Amazon EC2 with Hadoop and up to 100 High-CPU Medium instances, each providing two virtual cores.Each node was configured to run at most two map and reduce tasks in parallel.
  • Datasets: The evaluation uses DS1 with about 114,000 product descriptions and DS2 with about 1.4 million publication records.Both datasets use the first three title letters as the default blocking key, while robustness experiments vary blocking to study skew effects.

A. Robustness: Degree of data skew

Both proposed strategies remain effective under skewed block distributions and outperform Basic as parallelism increases, though their performance depends on redistribution overhead and input partitioning.

  • Robustness to data skew: For s=1, Basic requires 225 ms per 10^4 comparisons, more than 12 times slower than BlockSplit and PairRange.
  • Robustness to data skew: BlockSplit and PairRange remain stable across data skews, with PairRange holding a small advantage from more uniform workloads.
  • Number of reduce tasks: At r=160, BlockSplit and PairRange reduce execution time by a factor of 6 compared with Basic.
  • Number of reduce tasks: PairRange eventually outperforms BlockSplit by 7% as the number of reduce tasks increases, although it can be slower for small r.
  • Input partitioning: Sorting input by blocking key deteriorates BlockSplit execution time by 80% because large blocks become concentrated in the same map partitions.
  • Map-phase overhead: PairRange map output grows almost linearly with reduce-task count, whereas BlockSplit output changes in steps determined by input partitions and split blocks.

C. Scalability: Number of nodes

BlockSplit and PairRange distribute work across reduce tasks and nodes more effectively than Basic, scaling substantially farther on both datasets. Their relative advantage depends on dataset size, task granularity, and input partitioning.

  • Experimental setup: The evaluation uses node counts from 1 to 100, with m = 2·n map tasks and r = 10·n reduce tasks.
  • Scalability: Basic fails to scale beyond two nodes because each block is processed within a single reduce task.
  • Scalability: BlockSplit and PairRange scale almost linearly to 10 nodes for DS1 and 40 nodes for DS2.
  • Scalability: For large node counts, DS2 achieves better speedup than DS1 because its workload per reduce task is more suitable for efficient core utilization.
  • Strategy trade-offs: BlockSplit outperforms PairRange for DS1 at 100 nodes, where PairRange’s additional key-value-pair overhead harms execution time.
  • Strategy trade-offs: BlockSplit is preferable for smaller splittable datasets when data order is independent of the blocking key; otherwise, PairRange performs better.

APPENDIX I MATCHING TWO SOURCES

The two-source extension assumes valid blocking keys and adapts the matching workflow to distinguish entities from sources R and S. Entities without blocking keys are handled by adding Cartesian-product matches using a constant blocking key.

  • Entities without blocking keys: Entities without blocking keys are matched by combining regular valid-key matching with Cartesian products against the entities lacking keys.A constant blocking key ⊥ makes all required cross-source pairs eligible.
  • Input organization: The two-source setup assumes each input partition contains entities from only one source, although the sources may use different numbers of partitions.Hadoop’s MultipleInputs feature can enforce the source-specific partitioning assumption.
  • Input organization: In the running example, source R occupies one partition while source S spans two partitions and uses blocking keys w-z.The example entities are labeled A-N and distributed across partitions Π0, Π1, and Π2.
  • Source-aware matching: Block matching distinguishes the two sources by tagging blocks and entities with their source during BDM and map processing.The BDM identifies same-key blocks separately for R and S, while map outputs preserve source information.

A. Block-based Load Balancing

BlockSplit extends block-based load balancing to two sources by preserving source identity while splitting oversized blocks across reduce tasks. In the example, the largest block is divided so the 12 comparisons are assigned across three reducers.

  • BlockSplit strategy: BlockSplit enriches keys with source and split information, allowing reducers to identify entity pairs from different sources.Map outputs use keys containing the reduce index, block index, split, and source.
  • BlockSplit strategy: The two-source BlockSplit workflow follows the same scheme as the one-source strategy while adding source annotations to map outputs.Source tags enable the reduce phase to distinguish cross-source comparisons.
  • Example dataflow: 12 overall pairs yield an average workload of 4 pairs, while the largest block requires 6 pairs and is split into two match tasks.The split tasks are 3.0×1 and 3.0×2, and the resulting tasks are ordered by pair count.

B. Pair-based Load Balancing

PairRange balances two-source matching by enumerating cross-source pairs within each block and assigning contiguous pair ranges to reducers. Entities are replicated to every range containing one of their relevant pairs.

  • Pair enumeration: PairRange enumerates pairs between entities from R and S that share a blocking key, using a column-oriented pair index.The pair index combines within-block coordinates with an offset for preceding blocks.
  • Range construction: 12 pairs are divided into three ranges of size 4, while blocks lacking entities in source S are excluded.In the example, block Φ1 is omitted because source S has no entity with that blocking key.
  • Range construction: For an R entity, the relevant pairs span its row across S entities; for an S entity, they span its column across R entities.The map phase identifies the ranges covering those row- or column-wise pairs.
  • Data routing: Map output keys contain range, block, source, and entity-index fields, while partitioning uses only the range index.Sorting uses the complete key, and reducers compare entities grouped by block.
  • Data routing: Entity C from R belongs to ranges ℜ1 and ℜ2 and is therefore sent to the second and third reduce tasks.Its emitted keys are (1.3.R.0) and (2.3.R.0).

APPENDIX II

The appendix gives pseudocode and implementation details for BlockSplit, PairRange, and the preprocessing BDM computation. It also describes how intermediate outputs and workload metadata support the second MapReduce job.

  • MapReduce implementation: The first MapReduce job writes each entity with its blocking key as additional output for the second job.Prohibiting input-file splitting preserves the relevant partitioning for the second MapReduce job.
  • BDM use: Map tasks read the BDM at initialization, storing only prior-partition entity totals for blocking keys present in their input.The BDM could instead be stored in HBase to avoid memory limitations.
  • BDM functions: The BDM exposes block indexes, entity counts by block and partition, and the overall number of entity pairs.These functions provide the metadata needed for workload calculation and routing.
  • BlockSplit implementation: BlockSplit assigns match tasks by ordering them by descending comparison count and repeatedly selecting the reduce task with the fewest assigned comparisons.The pseudocode then adds each task’s comparisons to the selected reducer’s workload.
  • Reduce phase: Reducers compare entity pairs and append matching pairs to the final output.The implementation includes explicit comparison and buffering operations for emitted matches.
  • PairRange implementation: PairRange computes range indexes from pair indexes and the number of comparisons per reduce task, then routes output by range.The pseudocode separates repartitioning by range from sorting and grouping by block and entity index.
Loading 1108.1631v1…