Universitat HamburgFachbereich Informatik

Verteilte Systeme und Informationssysteme

Komplexe InformationssystemeSeminararbeit

Deduplikation:Blocking-Verfahren mit MapReduce

Betreuer: Professor Dr. Norbert Ritter

Dipl.-Inf. Fabian Panse

Vorgelegt von:

Erik WittMatrikelnr.: 6204671Heidmuhlenweg 90c25336 Elmshorn0witt@informatik.uni-hamburg.de

Inhaltsverzeichnis

1 Einleitung 11.1 Motivation . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 11.2 Aufbau der Arbeit . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 1

2 Deduplikation 22.1 Problematik . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 22.2 Blocking-Verfahren . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 2

2.2.1 Sorted Neighborhood Blocking . . . . . . . . . . . . . . . . . . . . 3

3 MapReduce 53.1 Programmiermodell . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 53.2 Funktionsweise . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 5

3.2.1 Ausführung . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 63.2.2 Beispiel . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 7

3.3 Fehlertoleranz . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 83.4 Erweiterungen . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 8

4 Deduplikation mit MapReduce 94.1 Generischer Ansatz . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 94.2 SN-Blocking mit MapReduce . . . . . . . . . . . . . . . . . . . . . . . . . . 10

4.2.1 Sorted Reduce Partitions . . . . . . . . . . . . . . . . . . . . . . . . 104.2.2 JobSN . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 124.2.3 RepSN . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 134.2.4 Multi-pass SN . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 15

4.3 Lastverteilung . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 154.4 Evaluation . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 16

5 Fazit 18

I

1 Einleitung

1.1 Motivation

Informationsintegration ist im Zeitalter von Big Data ein großes Thema. Große Men-gen von Daten werden an verschiedensten Stellen gesammelt, müssen integriert undan zentraler Stelle gespeichert werden, um die Informationen nutzbar zu machen. Einwichtiger Schritt bei der Verarbeitung dieser Daten aus verschiedenen Quellen ist dieDeduplikation. Sie identifiziert und entfernt Duplikate und verknüpft Daten, die sich aufeine gemeinsame Entität beziehen.

Bei der Menge von Daten, die heutzutage beispielsweise in Data-Warehouse-Systemengespeichert sind, ist Deduplizierung ein sehr aufwendiger Prozess, der sequentiell kaumumsetzbar ist. Steigenden Datenmengen wird mit schnelleren Rechnern und vor allemmit Parallelität begegnet. Cluster mit hunderten oder sogar tausenden Knoten könnengroße Datenmengen in kurzer Zeit verarbeiten, vorausgesetzt die Berechnung ist paralle-lisierbar.

Das MapReduce-Framework unterstützt die Parallelisierung von Berechnungen aufgroßen Datenmengen. Das Programmiermodell bietet eine simple Schnittstelle, mit dereine Vielzahl von Berechnungsvorschriften formuliert werden können. Die Implemen-tierung von MapReduce übernimmt die eigentliche Parallelisierung auf einem konfigu-rierbaren Cluster und handhabt viele der Probleme, die bei verteilten Berechnungenauftreten.

Die Bedürfnisse der Deduplikation und die Fähigkeiten des MapReduce-Frameworksscheinen gut zueinander zu passen. Diesbezüglich beschäftigen sich Kolb et al. [KTR10a]mit der Frage, wie Deduplikation mit Hilfe des MapReduce-Frameworks umgesetzt wer-den kann.

1.2 Aufbau der Arbeit

Diese Arbeit beschäftigt sich hauptsächlich mit Publikationen von Kolb et al. zu Dedupli-kation mit MapReduce und ist wie folgt aufgebaut. Kapitel 2 und 3 stellen die Grundlagenvon Deduplikation und dem MapReduce-Framework dar. In Kapitel 4 werden verschiede-ne Ansätze von Kolb et al. zur Deduplikation mit MapReduce [KTR10a, KTR12b] und zurLastverteilung [KTR11] vorgestellt. Kapitel 5 fasst die Arbeit zusammen und gibt einenAusblick.

1

2 Deduplikation

2.1 Problematik

Für eine oder mehrere gegebene Datenquellen ist Deduplikation das Problem, alle Ob-jekte zu identifizieren, die sich auf das gleiche realweltliche Objekt beziehen. Dedupli-kation ist ein wichtiger Schritt bei der Vorverarbeitung von Daten in vielen Bereichen.Es werden beispielsweise redundante Kundendaten entfernt, personenbezogene Datenaus mehreren Quellen verknüpft oder Angebote zu Preisvergleichen zusammengefasst.[KTR10a, KTR10b, Abschnitt 1]

Als Beispiel sei angenommen, es sollen Angebote zu verschiedenen Produkten ausmehreren Datenbanken integriert werden, um einen Preisvergleich zu erstellen. Einenaive Herangehensweise würde jedes Angebot mit jedem anderen vergleichen und dieAngebote zusammenfassen, wenn sie den gleichen Bezeichner haben oder alle Attribute,die das Produkt beschreiben, übereinstimmen. Dieses naive Vorgehen hat zwei essentielleProbleme [HS95, Abschnitt 2.1]:

1. Die resultierende Komplexität für die Vergleichsoperationen bei N Datenbankob-jekten ist O(N2). Diese quadratische Komplexität führt bei großen Datenbeständenzu nicht akzeptablen Laufzeiten. [KTR10a, Abschnitt 1]

2. Die Ausgangsdaten enthalten typischerweise keine eindeutigen Bezeichner und At-tributwerte sind häufig fehlerhaft bzw. unsauber, wodurch die Prüfung auf Gleich-heit aller Attribute fehlschlägt. [Chr12, Kapitel 1.1.1] Zur Erkennung von Objekten,die sich auf das gleiche realweltliche Objekt beziehen, werden typischerweise statis-tische Verfahren, logische Regeln oder empirisches Wissen über die Daten benötigt.

In dieser Arbeit wird nicht auf konkrete Techniken zum Vergleich von Objekten einge-gangen. Die vorgestellten Verfahren verringern und parallelisieren lediglich die nötigenVergleiche und sind unabhängig davon, wie die Objekte verglichen werden.

2.2 Blocking-Verfahren

Mit Blocking-Verfahren soll die Anzahl der Vergleiche verringert werden. Dazu werdendie Objekte anhand eines ihrer Attribute (oder anhand einer Teilmenge der Attribute)in Blöcke gruppiert. Das Kriterium für die Gruppierung wird Blocking-Key genannt. An-schließend werden die Objekte innerhalb jedes Blocks paarweise verglichen. [BCC03,Abschnitt 1] Abbildung 1 zeigt den generellen Ablauf von Deduplikation mit Blocking.

2

.

.

....

MatchResult

Matching

Matching

Matchingblock1

block1

block1

Blocking

Data

Source

Data

Source

Abbildung 1: Genereller Ablauf von Deduplikation mit Blocking.

Bei M überschneidungsfreien Blöcken und einer mittleren Blockgröße von NM verrin-

gert das Verfahren die Anzahl der Kandidaten von O(N2) auf O(M ·

(NM

)2)= O

(N2

M

).

Offensichtlich kann es bei diesem Verfahren dazu kommen, dass einige Duplikate nichterkannt werden, weil die entsprechenden Objekte in verschiedene Blöcke gruppiert wur-den und damit keine Kandidaten für einen Vergleich sind. Deshalb ist die Genauigkeitvon Deduplikation mit Blocking maßgeblich von der Güte des Blocking-Keys abhängig.

Im Allgemeinen teilt sich die Deduplikation mit Blocking damit in die Phasen Blocking(Gruppieren der Eingangsdaten) und Matching (paarweises Vergleichen innerhalb derBlöcke). Eine für diese Arbeit besonders relevante Variante der Blocking-Verfahren istdas Sorted Neighborhood Blocking, welches im folgenden Abschnitt beschrieben wird.

2.2.1 Sorted Neighborhood Blocking

Sorted Neighborhood Blocking (SN-Blocking) ist ein populäres Blocking-Verfahren, wel-ches in drei Schritten arbeitet [HS95, Abschnitt 2.2]:

1. Für jedes Objekt der Eingabedaten wird ein Blocking-Key basierend auf einer Teil-menge seiner Attribute berechnet.

2. Die Objekte werden nach ihrem Blocking-Key sortiert.

3. Zum Erzeugen der sich überschneidenden Blöcke wird ein Fenster fester Größeschrittweise über die Liste bewegt. Bei einer Fenstergröße w bildet jedes Objekt mitden folgenden w−1 Objekten einen Block. Um mehrfache Vergleiche zu verhindernbildet lediglich das erste Objekt eines Blocks mit jedem folgenden Element einenKandidaten für den Vergleich.

3

c - i

h - g

f - c

e - h

b - f

d - e

a - b

g - i

c - g

h - c

f - h

e - f

b - e

d - b

a - d

SKSKS

3

3

3

2

2

2

2

1

1

i

g

c

h

f

e

b

d

a

3

2

3

2

2

1

3

2

1

i

h

g

f

e

d

c

b

a

i

h

g

f

e

d

c

b

a

sortassigne blocking key

Abbildung 2: Beispiel für SN-Blocking.

SN-Blocking produziert(n− w

2

)· (w − 1) ∈ O(N · w) Kandidaten für einen Vergleich.

Dazu addiert sich die Komplexität zum Sortieren der Objekte von O(N · logN), was ineiner Komplexität von O(N ·w+N · logN) resultiert. [KTR10a, Abschnitt 4] Mit der Wahlder Fenstergröße w kann zwischen Komplexität und Genauigkeit abgewogen werden.

Abbildung 2 zeigt ein Beispiel für SN-Blocking, in dem den Objekten a, . . . , i zuersteine Blocking-Key zwischen 1 und 3 zugewiesen wird. Im zweiten Schritt werden dieObjekte nach diesem Schlüssel sortiert und anschließend wird ein Fenster der Größe 3

über die Liste bewegt (symbolisiert durch grüne Balken), um die Vergleichs-Kandidatenauf der rechten Seite der Abbildung zu generieren.

4

3 MapReduce

Bei MapReduce handelt es sich nach [DG08] um ein Programmiermodell mit einer zuge-hörigen Implementierung. Es wurde bei Google entwickelt und 2004 veröffentlicht. DasProgrammiermodell ermöglicht einfache Berechnungen über großen Datenbeständen par-allel auf den Knoten eines Clusters auszuführen. Die Implementierung übernimmt dabeidie Verteilung der Daten, die Ausführung der parallelen Berechnung, die Kommunikationim Cluster, die Fehlertoleranz und die Lastverteilung.

3.1 Programmiermodell

Ein sogenannter MapReduce-Job erhält eine Menge von Schlüssel-Wert-Paaren als Ein-gabe und berechnet wiederum eine Menge von Schlüssel-Wert-Paaren als Ausgabe. DieBerechnung wird durch den Benutzer mit Hilfe der Funktionen map und reduce spezifi-ziert. Diese Funktionen selbst sind sequentiell, werden aber parallel auf verschiedenenMaschinen ausgeführt. [DG08, KTR10a]

Jeder Aufruf der map-Funktion erhält eines der Schlüssel-Wert-Paare als Eingabe (in)und berechnet daraus eine Menge von temporären Schlüssel-Wert-Paaren (tmp):

map : (keyin, valuein)→ list(keytmp, valuetmp)

Die temporären Schlüssel-Wert-Paare werden nach dem Schlüssel gruppiert und an diereduce-Funktion weitergegeben.

Die reduce-Funktion erhält in jedem Aufruf einen der temporären Schlüssel und alleWerte, die zu diesem Schlüssel gehören, als Liste. Die Funktion berechnet hieraus dieAusgabe des MapReduce-Jobs als Liste von Schlüssel-Wert-Paaren (out).

reduce : (keytmp, list(valuetmp))→ list(keyout, valueout)

Alle Ausgaben der reduce-Funktionen bilden zusammen das Ergebnis des MapReduce-Jobs.

3.2 Funktionsweise

Die Ausführung der map-Funktion wird über mehrere Knoten verteilt, indem die Ein-gangsdaten automatisch in M Teilmengen partitioniert werden. Jeweils ein map-Prozessist für die Bearbeitung einer Partition zuständig. Die map-Prozesse können parallel aufmehreren Knoten ausgeführt werden, da diese Unabhängig voneinander sind.

5

Die Ausführung der reduce-Funktionen wird über mehrere Knoten verteilt, indem dieMenge der temporären Schlüssel mit Hilfe einer Partitionierungsfunktion in R Teilmen-gen partitioniert wird. Jeweils ein reduce-Prozess ist für die Bearbeitung einer Partitionzuständig. Die reduce-Prozesse können parallel auf mehreren Knoten ausgeführt werden.

Die Zahl der map-Prozesse M und der reduce-Prozesse R sowie die Partitionierungs-funktion können vom Nutzer bestimmt werden. Typischerweise sind M und R deutlichgrößer als die Anzahl der Knoten im Cluster, um dem System die Möglichkeit zu geben,die Last gleichmäßig auf die Knoten zu verteilen. [DG08]

3.2.1 Ausführung

Bei der Ausführung eines MapReduce-Jobs werden nach [DG08] die folgenden Aktionenausgeführt:

1. Das System teilt die Eingabedaten in M Partitionen und startet den Job auf denKnoten des Clusters.

2. Ein ausgewählter Knoten ist der sogenannte Master. Er verteilt die M map-Prozesseund R reduce-Prozesse auf inaktive Knoten.

3. Ein Knoten, dem ein map-Prozess zugewiesen ist, lädt die zugehörigen Eingangs-daten als Schlüssel-Wert-Paare. Mit jedem Paar wird die vom Benutzer definiertemap-Funktion aufgerufen. Die temporären Ausgaben der map-Funktion werden imArbeitsspeicher zwischengespeichert.

4. Die temporären Schlüssel-Wert-Paare werden regelmäßig aus dem Arbeitsspeichergelesen, mit der Partitionierungsfunktion partitioniert und in R Dateien auf dielokale Festplatte geschrieben. Die Adresse der Dateien wird dem Master zugesendet.

5. Der Master schickt die Adressen der temporären Daten an die jeweiligen reduce-Prozesse. Wenn ein Prozess alle seine Daten geladen hat, sortiert und gruppiert erdie Schlüssel-Wert-Paare anhand der Schlüssel.

6. Die reduce-Funktion wird mit jeder dieser Gruppen – also mit dem Schlüssel undder Menge der zugehörigen Werte – aufgerufen. Die Ausgabe der Funktion wirdpro reduce-Prozess in einer Datei abgelegt.

7. Wenn alle map- und reduce-Prozesse abgeschlossen sind, terminiert der MapReduce-Job. Das Ergebnis des Jobs ist jetzt über R Dateien verteilt. Die Daten können alsErgebnis zusammengeführt werden oder eine schon verteilte Eingabe für einenweiteren MapReduce-Job sein.

6

' for' 1

'go 1

1'lords'

tmptmp

'last' 2

3'the'

tmptmp

'we'

'of'

'time'

1

2

2

1'the'

1'the'

1'time'

1'the'

tmptmp

'we'

'of'

'of'

'time'

1

1

1

1

1'last'

'for' 1

'go 1

1'lords'

tmptmp

'last' 1

1'last'

1'the'

1'the'

1

'for' 1

'time'

1'the'

1'lords' PARTITI

ONING

ReduceMap

tmptmpIn

tmptmpIn

'go

'we'

'of'

'of'

'time'

'last'

1

1

1

1

1

1

c.txt

b. tx t

a.txt

reduce

reduce

map

map

Abbildung 3: MapReduce Beispiel zum Zählen von Wortvorkommen.

3.2.2 Beispiel

Ein verbreitetes Beispiel zur Erklärung von MapReduce ist das Zählen von Wortvorkom-men in Text-Dokumenten. [DG08, KTR10a, KTR12b] Die Eingabe ist eine Menge vonText-Dokumenten1. Die Ausgabe sind Paare aus dem jeweiligen Wort und der Anzahlseiner Vorkommen in allen Dokumenten. Das Beispiel ist in Abbildung 3 dargestellt.

Die drei Dokumente a.txt, b.txt und c.txt werden von zwei map-Prozessen verarbeitet.Dabei erhält jeder Aufruf der map-Funktion jeweils ein Dokument. Die map-Funktiongibt für jedes im Dokument vorkommende Wort w ein temporäres Schlüssel-Wert-Paar(w, 1) aus.

Die temporären Paare werden nach ihrem Schlüssel gruppiert und an die zwei reduce-Prozesse weitergegeben. Dabei erhält der erste Prozess alle Paare, deren Schlüssel mita−m beginnen, und der zweite Prozess die übrigen. Jeder Aufruf der reduce-Funktionerhält eine Gruppe von Paaren, also den Schlüssel und alle Werte, die zu diesem Schlüssel

1Bei der Eingabe handelt es sich um Schlüssel-Wert-Paare, bei denen der nicht relevante Schlüssel wegge-lassen wurde.

7

gehören. Im Falle des zweiten reduce-Prozesses und dem Schlüssel ‘the‘ ist der Aufrufreduce(‘the‘, [1, 1, 1]). Die reduce-Funktion addiert lediglich die Werte und gibt das Er-gebnis zusammen mit dem Schlüssel aus.

3.3 Fehlertoleranz

Bei der Parallelisierung von Berechnungen auf großen Clustern ist Fehlertoleranz beson-ders wichtig, weil die Ausfallwahrscheinlichkeit von einzelnen Knoten mit der Größedes Clusters steigt. Das MapReduce-System setzt diese Eigenschaft ohne Einwirken desBenutzers um. [DG08]

Der ausgewählte Knoten (Master) überprüft die Verfügbarkeit aller anderen Knotenüber regelmäßige ping-Nachrichten. Wenn ein Knoten nicht erreichbar ist, wird er alsausgefallen markiert. Alle auf dem Knoten aktiven map- oder reduce-Prozesse werdenzurückgesetzt und durch den Master an andere Knoten vergeben. Bereits abgeschlosse-ne map-Prozesse müssen ebenfalls zurückgesetzt und neu vergeben werden, weil dietemporären Daten auf der Festplatte des ausgefallenen Knotens liegen. Abgeschlossenereduce-Prozesse müssen nicht wiederholt werden2.

Wenn die map- und reduce-Funktionen des Benutzers deterministisch sind, entsprichtdas Ergebnis des MapReduce-Jobs nach dem Behandeln des Fehlerfalls dem einer fehler-freien sequenziellen Ausführung des Programms.

3.4 Erweiterungen

Die Basisfunktionalität von MapReduce wurde in verschiedener Hinsicht erweitert, so-wohl in der Veröffentlichung von Google selbst [DG08] als auch in der weit verbreitetenImplementierung Hadoop [Apa15a]. An dieser Stelle werden nur die für diese Arbeitrelevanten Erweiterungen beschrieben.

Im Unterabschnitt 3.2 wurde schon darauf hingewiesen, dass der Benutzer eine eigenePartitionierungsfunktion angeben und damit bestimmen kann, welche Teilmenge derSchlüssel ein reduce-Prozess erhalten soll. Ebenfalls als Funktion über den Schlüsselnkann der Nutzer in der Eweiterung angeben, wie die temporären Schlüssel-Wert-Paarepro reduce-Prozess sortiert werden sollen. Es kann sogar spezifiziert werden, wie dieSchlüssel für die reduce-Aufrufe zu gruppieren sind. [Apa15b] Diese Erweiterungenwerden im Unterunterabschnitt 4.2.1 verwendet und sind essentiell für die Umsetzungvon SN-Blocking mit MapReduce.

2Der Ausfall des Masters wird zumindest in der Implementierung von Google nicht toleriert, weil diesesEreignis im Allgemeinen unwahrscheinlich ist. Es gibt hierfür allerdings Lösungsansätze.

8

4 Deduplikation mit MapReduce

4.1 Generischer Ansatz

Die Umsetzung der allgemeinen Deduplikation mit Blocking mit Hilfe von MapReduce istsimpel. Die Blocking-Phase wird mit der map-Funktion und die Matching-Phase mit derreduce-Funktion umgesetzt. Dabei wird in der map-Funktion der Blocking-Key für dieObjekte berechnet, das MapReduce-System gruppiert die Objekte nach ihrem Blocking-Key und verteilt die Gruppen an die reduce-Prozesse. In der reduce-Funktion werden alleObjekte (einer Gruppe) paarweise miteinander verglichen [KTR10a, Abschnitt 3]

Abbildung 4 zeigt ein Beispiel dieses generischen Ansatzes. Die Menge der Einga-beobjekte a, . . . , i werden auf drei map-Prozesse aufgeteilt, in denen den Objekten je-weils ein Blocking-Key zwischen 1 und 3 zugeordnet wird. Das System gruppiert dietemporären Schlüssel-Wert-Paare und verteilt die drei entstehenden Gruppen an zweireduce-Prozesse. Mit den grünen Balken ist angedeutet, welche Vergleiche im jeweiligenreduce-Aufruf ausgeführt werden.

Dieser generische Ansatz stellt eine simple Parallelisierung von Deduplikation mitBlocking dar, hat nach [KTR10a, Abschnitt 3] allerdings auch eine Reihe von Problemen:

• Disjunkte Partitionierung: Im MapReduce-System werden die temporären Schlüssel-Wert-Paare anhand ihres Schlüssels jeweils an genau einen reduce-Prozess gegeben.Dies erschwert die Umsetzung von Blocking-Verfahren mit überschneidenden Blö-cken, wie beispielsweise SN-Blocking.

• Lastverteilung: Je nach Blocking-Key und Beschaffenheit der Eingangsdaten kannes dazu kommen, dass die Blockgrößen sehr ungleich verteilt sind. Das führt da-zu, dass einige reduce-Prozesse deutlich mehr Objekte erhalten und damit mehrRechenzeit benötigen als andere. Im schlimmsten Fall kann das parallele Verfahrenzu einem sequenziellen degenerieren.

• Speicherengpässe: Die reduce-Aufrufe erhalten alle Objekte aus einem Block inForm eines Iterators und können dementsprechend auf die Elemente nur striktnacheinander zugreifen. Die Funktion soll allerdings jedes Element mit jedem an-deren vergleichen und muss daher erst einmal alle Elemente in den Speicher laden.Bei großen Datenmengen kann es dabei zu Speicherengpässen kommen. DiesesProblem wird durch das Problem der Lastverteilung noch verschärft.

Im folgenden Abschnitt wird das Problem der disjunkten Partitionierung aufgegriffen,indem zwei Verfahren zur Umsetzung von SN-Blocking vorgestellt werden. Diese Ver-fahren adressieren auch das Problem der Speicherengpässe, weil nicht alle Elemente

9

K

K

3

3

3

2

2

2

2

1

1

PARTITI

ONING

Reduce-Phase = MatchingMap-Phase = Blocking

S

S

i

g

c

h

f

e

b

d

a

SK

SK

S

S

c - i

e - h

b - f

g - i

c - g

f - h

e - f

b - e

a - d

SKS

3

2

3

2

2

1

3

2

1

i

h

g

f

e

d

c

b

a

i

h

g

f

e

d

c

b

a

assigne blocking nr.

assigne blocking nr.

assigne blocking nr.

Abbildung 4: Deduplikation mit generischen Blocking-Verfahren.

eines Blocks paarweise verglichen werden, sondern nur die Elemente im angesprochenenFenster. Der Unterabschnitt 4.3 beschäftigt sich mit dem Problem der Lastverteilung.

4.2 SN-Blocking mit MapReduce

In diesem Kapitel werden zwei Verfahren zur Umsetzung von SN-Blocking mit MapRedu-ce vorgestellt, die beide auf dem Sorted Reduce Partitions Verfahren basieren.

4.2.1 Sorted Reduce Partitions

Das Sorted Reduce Partitions (SRP) Verfahren setzt einen Teil des SN-Blockings alsMapReduce-Job um. Dabei wird der Blocking-Key von der map-Funktion und die Ver-gleichskandidaten in der reduce-Funktion berechnet. Wichtig ist dabei, dass die reduce-Prozesse die Objekte sortiert nach ihrem Blocking-Key erhalten.

Das SRP-Verfahren nach [KTR10a, Abschnitt 4.1] erreicht diese Voraussetzung durcheine Kombination aus angepassten Partitionierungs-, Sortier- und Gruppierungsfunktio-

10

PARTITI

ONING

missing!

Reduce-PhaseMap-Phase

h - gh - c

f - c

SK

2.3

2.3

2.3

i

g

c

SK

SK

2.3

2.3

2.3

1.2

1.2

1.2

1.2

1.1

1.1

i

g

c

h

f

e

b

d

a

SK

SK

S

S

c - i

e - h

b - f

d - e

a - b

g - i

c - g

f - h

e - f

b - e

d - b

a - d

SKSKS

1.2

1.2

1.2

1.1

1.2

1.1

h

f

e

d

b

a

2.3

1.2

2.3

1.2

1.2

1.1

2.3

1.2

1.1

i

h

g

f

e

d

c

b

a

i

h

g

f

e

d

c

b

a

sort

sortassigne blocking nr.

assigne blocking nr.

assigne blocking nr.

Abbildung 5: Beispiel für Sorted Reduce Partitions.

nen. Diese Funktionen basieren auf einem zusammengesetzten Schlüssel K = i.k, dervon der map-Funktion berechnet wird. Die erste Komponente i ist eine Zahl zwischen 1

und der Anzahl der reduce-Prozesse R. Die zweite Komponente k ist der Blocking-Keydes Objekts. Die Partitionierungsfunktion verteilt alle Objekte mit erster Komponente i anden i-ten reduce-Prozess. Dabei muss gelten, dass alle Objekte des i-ten reduce-ProzessBlocking-Keys besitzen, die kleiner oder gleich der Objekte des i+ 1-ten reduce-Prozesssind3.

Die Sortierfunktion ordnet die Objekte pro reduce-Prozess nach der zweiten Komponen-te k, also dem Blocking-Key. Die Gruppierungsfunktion wiederum gruppiert die Objektenach der ersten Komponente i. Damit werden alle Objekte eines reduce-Prozesses ineinem reduce-Aufruf bearbeitet. Dies ist notwendig, um das Fenster über die Liste allersortierten Werte zu bewegen.

Abbildung 5 zeigt das Verfahren an einem Beispiel. Drei map-Prozesse berechnen denSchlüssel K = i.k für die Eingabeobjekte a, . . . , i. Die erste Komponente i identifiziertden zuständigen reduce-Prozess und wird von der Partitionierungsfunktion verwendet.Die zweite Komponente k ist der Blocking-Key, der zwischen 1 und 3 liegt und nach derPartitionierung zum Sortieren der Objekte verwendet wird. Das Partitionieren, Sortierenund Gruppieren wird vom MapReduce-System mit Hilfe der vom Benutzer spezifizier-

3Wenn die Wertemenge der Blocking-Keys bekannt ist, kann diese Voraussetzung mit einer einfachenBereichspartitionierung umgesetzt werden.

11

ten Funktionen übernommen. Die zwei reduce-Prozesse bilden jeweils mit Hilfe einesFensters der Größe 3 (symbolisiert durch grüne Balken) die Vergleichs-Kandidaten.

Wie auch im Beispiel zu sehen ist generiert dieses Verfahren im Allgemeinen nichtalle Vergleichs-Kandidaten des SN-Blocking (vergleiche mit Unterunterabschnitt 2.2.1),weil die Überlappung der Daten jeweils zwischen den reduce-Prozessen i und i + 1

nicht berücksichtigt werden kann. Die folgenden Verfahren JobSN und RepSN setzen aufdiesem Verfahren auf und beheben die angesprochene Unzulänglichkeit.

4.2.2 JobSN

Das JobSN Verfahren nach [KTR10a, Abschnitt 4.2] baut auf dem SRP-Verfahren aufund verwendet einen weiteren MapReduce-Job, um die fehlenden Vergleichskandidatenzu generieren und damit das SN-Blocking korrekt zu implementieren. Bei den fehlen-den Vergleichskandidaten handelt es sich um die Paarungen zwischen den letzen w − 1

Objekten des i-ten reduce-Prozesses und den ersten w − 1 Objekten des i+ 1-ten reduce-Prozesses. Um diese Paarungen zu generieren, gibt jeder reduce-Prozess die ersten unddie letzten w − 1 Schlüssel-Wert-Paare als Eingabe an einen weiteren MapReduce-Job.Der erste bzw. der letzte Prozess gibt hierbei lediglich seine letzten bzw. ersten w − 1

Schlüssel-Wert-Paare weiter.

Vor dem Übergeben der Werte wird jedem Schlüssel-Wert-Paar eine weitere Schlüssel-komponente j zugeordnet, sodass ein Schlüssel K = j.i.k entsteht. Die Komponente j

ist gleich i für die w − 1 letzten Objekte des i-ten reduce-Prozesses und ebenfalls für diew − 1 ersten Objekte des i+ 1-ten reduce-Prozesses.

Der zweite MapReduce-Job lässt die Eingangswerte in der map-Funktion unverändert.Die Partitionierung entspricht der Komponente j, die Sortierung wieder dem Blocking-Keyk und die Gruppierung wiederum der Komponente j. Damit erhält der reduce-Aufruf desi-ten reduce-Prozesses genau die Objekte zur Generierung der Paarungen, die zwischenProzess i und i+ 1 des ersten MapReduce-Jobs fehlten.

Der reduce-Prozess des zweiten Jobs generiert die Paarungen mit einem Fenster dergleichen Größe. Objekte, die sich in der Schlüsselkomponente i nicht unterscheiden,werden übersprungen, weil diese Paare bereits im ersten Job generiert wurden.

Abbildung 6 zeigt ein Beispiel für das JobSN Verfahren. Der linke Teil stellt den erstenMapReduce-Job dar und entspricht dem Beispiel aus Unterunterabschnitt 4.2.1. Der erstereduce-Prozess des ersten Jobs gibt seine letzten 3 − 1 = 2 Objekte f und h mit derSchlüsselkomponente j = 1 an den zweiten Job weiter. Der zweite reduce-Prozess tutdasselbe mit den ersten beiden Objekten c und g. Die map-Funktion des zweiten Jobs lässt

12

K

1.2.3

1.2.3

1.1.2

1.2K

clearedduplicates

PARTITI

ONING

AND

SORT

missing!

h - g

f - c

c - g

h - c

f - h

PARTITI

ONING

AND

SORT

1.2.3

1.2.3

g

c

SK

1.1.2

1.1.2

h

f

Reduce-Phase

Map-Phase

SK

1.2.3

1.2.3

g

c

SK

1.1.2

1.1.2

h

f

g

c

S

h

f

S

c,g

f, h

Reduce-Phase

Map-Phase

h - gh - c

f - c

SK

SK

2.3

2.3

2.3

1.1.2

1.2

1.2

1.2

1.1

1.1

i

g

c

h

f

e

b

d

a

SK

SK

S

S

c - i

e - h

b - f

d - e

a - b

g - i

c - g

f - h

e - f

b - e

d - b

a - d

SKS

2.3

1.2

2.3

1.2

1.2

1.1

2.3

1.2

1.1

i

h

g

f

e

d

c

b

a

i

h

g

f

e

d

c

b

a

no changes

no changes

assigne blocking nr.

assigne blocking nr.

assigne blocking nr.

Abbildung 6: Beispiel für das JobSN-Verfahren.

die Schlüssel-Wert-Paare unverändert und das System übergibt die sortierten Objekte aneinen reduce-Prozess, der im einzigen reduce-Aufruf die fehlenden, in grün dargestelltenPaarungen berechnet und dabei die rot markierten Paarungen überspringt.

4.2.3 RepSN

Das RepSN Verfahren nach [KTR10a, Abschnitt 4.3] basiert auch auf dem SRP-Verfahren,aber verwendet im Gegensatz zum JobSN Verfahren nur einen MapReduce-Job. Damitauch Paarungen zwischen den Objekten zweier aufeinanderfolgender reduce-Prozesseberücksichtig werden, muss die folgende Bedingung gelten:

(1) Der i-te reduce-Prozess muss zusätzlich mindestens die w − 1 letztens Objekte desi− 1-ten Prozesses erhalten.

Für das RepSN Verfahren wird die map-Funktion des SRP-Verfahrens erweitert. DerSchlüssel K = i.k aus dem SRP-Verfahren wird um eine Komponente j zu K = i.j.k er-weitert. Für jedes Schlüssel-Wert-Paar der map-Funktion des SRP-Verfahrens mit Schlüsseli.k generiert die neue map-Funktion ein Schlüssel-Wert-Paar mit Schlüssel i.i.k.

Um die beschriebene Voraussetzung (1) zu erfüllen, generiert die map-Funktion zu-sätzlich für alle reduce-Prozesse i, bis auf den letzten, Schlüssel-Wert-Paare mit Schlüsseli + 1.i.k, für die w − 1 Schlüssel i.i.k mit maximalem k. Dies stellt sicher, dass der i-tereduce-Prozess zusätzlich mindestens die w − 1 letzten Objekte (mit maximalem k) desi− 1-ten Prozesses erhält. Da jeder map-Prozess bis zu w − 1 Objekte repliziert, entstehtim Allgemeinen eine Menge unnötiger Schlüssel-Wert-Paare. Weil die map-Prozesse nurInformationen über ihre eigenen Daten haben, können diese unnötigen Replikate nichteliminiert werden.

13

clearedduplicates

g - i

c - ic - g

h - g

f - c

e - h

b - f

a -e

h - c

f - h

e - f

b - e

a - b

2.1.2

2.1.2

2.1.2

2.1.2 b

e

f

h

a2.1.1

h

f

e

b

a

2.1.2

2.1.2

2.1.2

2.1.2

2.1.1

h2.1.2

f2.1.2

e2.1.2

b2.1.2

a2.1.1

PARTITI

ONING

missing!

Reduce-PhaseMap-Phase

h - gh - c

f - c

SK

2.2.3

2.2.3

2.2.3

i

g

c

SK

SK

2.2.3

2.2.3

2.2.3

1.1.2

1.1.2

1.1.2

1.1.2

1.1.1

1.1.1

i

g

c

h

f

e

b

d

a

SK

SK

S

S

e - h

b - f

d - e

a - b

f - h

e - f

b - e

d - b

a - d

SKSKS

1.1.2

1.1.2

1.1.2

1.1.1

1.1.2

1.1.1

h

f

e

d

b

a

2.2.3

1.1.2

2.2.3

1.1.2

1.1.2

1.1.1

2.2.3

1.1.2

1.1.1

i

h

g

f

e

d

c

b

a

i

h

g

f

e

d

c

b

a

sort

sort

assigne blocking nr.

assigne blocking nr.

assigne blocking nr.

Abbildung 7: Beispiel für das RepSN-Verfahren.

Die Partitionierung der Schlüssel-Wert-Paare ergibt sich durch die erste Schlüsselkom-ponente i. Sortiert wird nach dem Blocking-Key k und gruppiert wiederum anhand derersten Komponente i. Die reduce-Funktion berechnet die Vergleichs-Kandidaten mit Hilfedes Fensters fester Größe. Paarungen, bei denen für beide Objekte i 6= j gilt, werdenübersprungen, weil diese Paarungen bereits im vorherigen reduce-Prozess berechnetwird.

Abbildung 7 zeigt das Beispiel vom SRP-Verfahren (Abbildung 5) nun mit dem RepSNVerfahren. Die map-Funktion produziert in diesem Beispiel die rot markierten Schlüssel-Wert-Paare zusätzlich zu denen aus dem SRP-Beispiel und repliziert damit unter anderemauch die letzten 3 − 1 = 2 Objekte f und h des ersten reduce-Prozesse in den zweitenreduce-Prozess. Zusätzlich werden die unnötigen Objekte a, b und e repliziert. Nachder Partitionierung, Sortierung und Gruppierung berechnen die reduce-Prozesse die er-warteten Paarungen. Der zweite reduce-Prozess überspringt dabei die rot markiertendurchgestrichenen Paare und berechnet unter anderem die grün markierten – im SRP-Beispiel fehlenden – Paare.

14

4.2.4 Multi-pass SN

Wie schon in Unterabschnitt 2.2 erwähnt, hängt die Genauigkeit von Blocking-Verfahrenmaßgeblich von der Zusammensetzung des Blocking-Keys ab. Trotz einstellbarer Fenster-größe gilt dies auch für SN-Blocking. Im Allgemeinen genügt ein einzelner Blocking-Keynicht, um alle Duplikate eines Datensatzes zu finden. [KTR12b, Abschnitt 4.4] [HS95,Abschnitt 2.4]

Multi-pass SN adressiert dieses Problem, indem SN-Blocking mehrfach mit verschie-denen Blocking-Keys ausgeführt und die Ergebnisse kombiniert werden. Typischerweisewird zusätzlich die transitive Hülle über den erkannten Duplikaten gebildet. [HS95, Ab-schnitt 2.4] Wenn also jeweils die Objekte a und b sowie b und c als Duplikate erkanntwurden, dann sind auch a und c Duplikate. In [KTR12b] wird das Verfahren MultiRepSNvorgestellt, in dem mehrere Ausführungen von RepSN in einem MapReduce-Job zusam-mengefasst werden.

4.3 Lastverteilung

Ein Problem bei der Umsetzung von Blocking-Verfahren mit MapReduce ist die Lastver-teilung (siehe Unterabschnitt 4.1). Je nach Blocking-Key und Eingabedaten kann es dazukommen, dass sich ein Großteil der Daten in wenigen Blöcken sammelt und damit einigereduce-Prozesse deutlich mehr Berechnungen ausführen als andere. Der Grund dafür ist,dass die Objekte strikt anhand ihres Blocking-Key partitioniert werden.

Bei der Umsetzung von SN-Blocking gibt es bei der Partitionierung größere Freiheits-grade. Die Partitionierung muss nur sicherstellen, dass die Schlüssel vom reduce-Prozessi kleiner oder gleich der Schlüssel vom reduce-Prozess i + 1 sind. Für die vorgestelltenVerfahren ist es also möglich, eine optimale Partitionierung der Objekte zu erreichen.

In den Veröffentlichungen [KTR12b] und [KTR11] beschäftigen sich die Autoren damit,eine möglichst uniforme Partitionierung der Objekte für die reduce-Phase zu erreichen.Dazu berechnen sie in einem vorgelagerten zusätzlichen MapReduce-Job die Häufig-keiten der verschiedenen Blocking-Keys in den Partitionen der Eingangsdaten. Anhanddieser Häufigkeiten kann eine annähernd uniforme Partitionierung berechnet werden.Das Verfahren ist sowohl auf JobSN und RepSN als auch auf MultiRepSN anwendbar.Im Fall von RepSN und MultiRepSN können mit Hilfe der Häufigkeiten sogar die inUnterunterabschnitt 4.2.3 angesprochenen, unnötigen Replikate eliminiert werden.

15

Abbildung 8: Vergleich der SN-Blocking Implementationen. [KTR10a, Abschnitt 5.2]

4.4 Evaluation

Die beiden Verfahren JobSN und RepSN wurden in [KTR10a, Abschnitt 5] in mehre-ren Experimenten auf ihre Laufzeit und ihren Speedup hin untersucht. Die Experimen-te wurden auf einem Cluster mit vier Knoten mit jeweils zwei Kernen ausgeführt. Eskonnten also maximal acht parallele map- bzw. reduce-Prozesse ausgeführt werden. AlsMapReduce-Implementierung wurde Hadoop in Version 0.20.2 verwendet. Die Testdatenwaren 1,4 Millionen Publikations-Datensätze.

Abbildung 8 fasst die Ergebnisse der Experimente zusammen. Das linke Diagrammzeigt die Laufzeit und den Speedup bei einer Fenstergröße von w = 10, das rechteDiagramm bei einer Fenstergröße von w = 1000. In beiden Diagrammen zeigt die x-Achse die Anzahl der parallel genutzten map- und reduce-Prozesse. Die Experimentewurden danach mit einem Prozess (sequentielle Ausführung), zwei, vier, sechs und achtparallelen Prozessen durchgeführt.

Im linken Diagramm ist zu sehen, dass RepSN mit nur einem MapReduce-Job etwaseffizienter arbeitet als JobSN. Bei einer Fenstergröße von w = 1000 ist dieser Unter-schied geringer. Im rechten Diagramm ist auch zu sehen, dass der Speedup nahezu linearund die Parallelisierung damit nahezu optimal ist. Bei der geringeren Fenstergröße isthingegen das Verhältnis von Berechnungsdauer einzelner Prozesse und dem Aufwandzum Verteilen der Daten schlechter und damit auch der Speedup geringer. Die im rech-ten Diagramm nicht ganz linearen Speedups für sechs und acht Prozesse erklären dieAutoren mit der Implementierung von MapReduce und vor allem mit den komplexenFehlertoleranzmechanismen.

In den Veröffentlichungen [KTR12b] und [KTR11] werden auch das Verfahren Multi-RepSN und die automatische Partitionierung zur Lastverteilung evaluiert. Abbildung 9zeigt im linken Diagramm einen Vergleich der Laufzeiten von RepSN mit einer einfa-

16

Abbildung 9: Vergleich RepSN und RepSN mit Lastverteilung (links). Vergleich der Qua-litätswerte von RepSN und MultiRepSN (rechts). [KTR12b, Abschnitt 6]

chen manuellen Partitionierung und RepSN mit der automatischen Partitionierung, dieim Unterabschnitt 4.3 beschrieben ist. Auf der x-Achse ist die Rate der Gleichverteilungder Daten auf die Blöcke zu sehen. Wobei 0 die Gleichverteilung und 1 eine maximalungleiche Verteilung repräsentiert. Es ist zu sehen, dass das manuelle RepSN Verfahrennur für nahezu gleich verteilte Daten effizienter ist. Bei einer ungleichen Verteilung –die bei Datensätzen in der Praxis zu erwarten ist – ist das Verfahren mit automatischerPartitionierung deutlich effizienter, weil es unabhängig von der Verteilung der Daten einekonstante Laufzeit aufweist.

Die Tabelle auf der rechten Seite von Abbildung 9 zeigt einen Vergleich zwischenRapSN und MultiRepSN (mit zwei unterschiedlichen Blocking-Keys). Zum Vergleich derEffizienz werden die Fenstergrößen, die Anzahl und die Reduktionsrate der Vergleichesowie die Laufzeit angegeben. Zum Vergleich der Qualität sind Precision, Recall und F-Measure angegeben. Aus der Tabelle wird ersichtlich, dass bei gleichzeitiger Reduktionder Laufzeit mit MultiRepSN eine ähnliche bzw. zum Teil bessere Qualität erreicht werdenkann. Dieses Erkenntnis stimmt auch mit den Ergebnissen in [HS95] überein.

17

5 Fazit

Zusammenfassend kann festgehalten werden, dass sich MapReduce sehr gut zur Umset-zung von Deduplikation eignet. Zum einen erfordert das Problem der Deduplizierung vongroßen Datenbeständen eine parallele Berechnung, um zu vertretbaren Laufzeiten zugelangen. Zum anderen lassen sich insbesondere Blocking-Verfahren gut mit MapReduce-Jobs umsetzten.

Die Arbeit hat dargestellt, dass die Probleme der disjunkten Partitionierung, der Last-verteilung und der Speicherengpässe für SN-Blocking gelöst werden können und dassinsbesondere das MultiRepSN Verfahren mit automatischer Partitionierung eine effizienteUmsetzung von Deduplikation auf großen Datensätzen ist.

Interessant – konkret im Bereich der Parallelisierung von SN-Blocking mit MapReduce –könnte eine weitere Untersuchung sein, die das Verhalten der behandelten Verfahren aufgroßen Clustern untersucht und feststellt, ob eine nahezu lineare Skalierung beibehaltenwerden kann. Bei Google bestehen MapReduce-Cluster typischerweise aus hundertenoder sogar tausenden von Knoten. [DG08]

Zur Verwendung der behandelten Verfahren in der Praxis haben Kolb et al. mit De-doop [KTR12a] ein Werkzeug entwickelt, mit dem - eigenen Angaben zufolge - komplexeDeduplikations-Prozesse auf großen Datensätzen spezifiziert und ausgeführt werden kön-nen. Damit dürfte auch die tatsächliche Anwendung der beschriebenen Techniken aufdie verschiedenen Anwendungsbereiche von Deduplikation möglich sein.

18

Literatur

[Apa15a] Apache. Apache hadoop website, 2015. Accessed: 2015-10-23.

[Apa15b] Apache. Interface reducer, 2015. Accessed: 2015-10-23.

[BCC03] Rohan Baxter, Peter Christen, and Tim Churches. A comparison of fast blockingmethods for record linkage. In ACM SIGKDD, volume 3, pages 25–27. Citeseer,2003.

[Chr12] P. Christen. Data Matching: Concepts and Techniques for Record Linkage, EntityResolution, and Duplicate Detection. Data-Centric Systems and Applications.Springer, 2012.

[DG08] Jeffrey Dean and Sanjay Ghemawat. Mapreduce: simplified data processingon large clusters. Communications of the ACM, 51:107–113, 2008.

[HS95] Mauricio A Hernández and Salvatore J Stolfo. The merge/purge problem forlarge databases. In ACM SIGMOD Record, volume 24, pages 127–138. ACM,1995.

[KTR10a] Lars Kolb, Andreas Thor, and Erhard Rahm. Parallel sorted neighborhoodblocking with mapreduce. arXiv preprint arXiv:1010.3053, 2010.

[KTR10b] Hanna Köpcke, Andreas Thor, and Erhard Rahm. Evaluation of entity reso-lution approaches on real-world match problems. Proceedings of the VLDBEndowment, 3:484–493, 2010.

[KTR11] Lars Kolb, Andreas Thor, and Erhard Rahm. Block-based load balancing forentity resolution with mapreduce. In Proceedings of the 20th ACM internatio-nal conference on Information and knowledge management, pages 2397–2400.ACM, 2011.

[KTR12a] Lars Kolb, Andreas Thor, and Erhard Rahm. Dedoop: efficient deduplicationwith hadoop. Proceedings of the VLDB Endowment, 5:1878–1881, 2012.

[KTR12b] Lars Kolb, Andreas Thor, and Erhard Rahm. Multi-pass sorted neighborhoodblocking with mapreduce. Computer Science-Research and Development, 27:45–63, 2012.

19