18.3 Parallele Joins

18.3.3 Parallele Hash-Joins

Hash-Joins gestatten eine effiziente Bearbeitung von Equi-Joins, da sie nur einen geringen CPU-Aufwand erfordern und große Hauptspeicher zur E/A-Reduzierung nutzen können. Wir beschreiben zunächst verschiedene Varianten der sequentiellen Hash-Join-Bearbeitung, für die dann eine parallele Realisierung für Shared-Nothing angegeben wird.

Sequentielle Realisierungen

Das Basisverfahren des sequentiellen Hash-Joins ist anwendbar, wenn die kleinere (innere) Relation S - nach Anwendung möglicher Selektionen und Projektionen - vollständig im Hauptspeicher gehalten werden kann. In diesem Fall besteht der Hash-Join aus zwei Phasen. In Phase 1 (building phase) wird die innere Relation S in eine Hash-Tabelle im Hauptspeicher gebracht, wobei eine Hash-Funktion h auf das Join-Attribut angewendet wird. Damit wird jedes Tupel von S einer bestimmten Hash-Klasse zugeordnet. In Phase 2 (probing phase) wird die zweite (äußere) Relation R gelesen. Für jeden R-Satz wird dabei wieder die Hash-Funktion h auf das Join-Attribut angewendet, und es wird überprüft, ob in der somit ermittelten Hash-Klasse S-Tupel mit übereinstimmendem Join-Attributwert vorliegen. Diese Bestimmung der Verbundpartner erfordert nur den Vergleich mit den S-Tupeln einer Hash-Klasse, so daß bei einer großen Anzahl von Hash-Klassen der Suchraum signifikant eingeschränkt wird. Die Kosten der Join-Berechnung sind damit sehr gering und linear zur Relationengröße. Die Bearbeitungsdauer ist im wesentlichen durch das Einlesen der beiden Relationen bestimmt.

Dieses optimale Leistungsverhalten ist jedoch nicht mehr erreichbar, wenn die innere Relation zu groß für eine vollständige Hauptspeicherallokation ist. In diesem Fall ist eine Überlaufbehandlung erforderlich, wofür es zahlreiche Alternativen gibt. Wir beschränken unsere Betrachtungen auf Ansätze, welche auf einer Partitionierung der Relationen basieren. Diese Idee wurde zuerst für die GRACE-Datenbankmaschine realisiert [Ki83]. Beim GRACE-Hash-Join wird vor der eigentlichen Join-Bearbeitung eine Partitionierungs-Phase durchgeführt (Abb. 18-4a). In dieser Phase wird zunächst die innere Relation S gelesen und mittels einer auf dem Join-Attribut angewendeten Partitionierungsfunktion g (Hash- oder Bereichspartitionierung) in q Partitionen S1 bis Sq unterteilt, so daß jede S-Partition im Hauptspeicher gehalten werden kann. Zur Partitionierung werden im Hauptspeicher q Ausgabepuffer von mindestens 1 Seite reserviert, welche auf Externspeicher ausgeschrieben werden, sobald sie keine weiteren S-Tupel mehr aufnehmen können. Das Ausschreiben der erzeugten Partitionen auf temporäre Dateien erfolgt somit weitgehend asynchron. Nach Partitionierung der S-Relation erfolgt eine analoge Zerlegung von R in q Partitionen R1 bis Rq unter Anwendung derselben Partitionierungsfunktion. Die Partitionierung garantiert, daß die Verbundpartner von Partition Si nur in der Partition Ri vorliegen können (für alle i, 1 i q). Daher genügt es zur eigentlichen Join-Berechnung, wie in Abb. 18-4b,c illustriert, das Hash-Join-Basisverfahren auf jede der q zusammengehörigen S- und R-Partitionen anzuwenden (Einlesen von Si in eine Hauptspeicher-Hash-Tabelle; Lesen von Ri mit Bestimmung der Verbundpartner).

Abb. 18-4: GRACE-Hash-Join

Eine Verbesserung des GRACE-Joins stellt der sogenannte Hybrid-Hash-Join dar [De84]. Er zeichnet sich durch eine bessere Hauptspeichernutzung während der Partionierungs-Phase aus. Dazu wird der von den Pufferbereichen nicht benötigte Hauptspeicherplatz bereits zur Allokation der Hash-Tabelle für die erste Partition S1 genutzt. Die Join-Berechnung für die erste Partition kann somit bereits beim ersten Lesen der Relationen S und R erfolgen, so daß für diese Partition keine zusätzlichen E/A-Vorgänge anfallen. Die Partitionierungsfunktion sollte so gewählt werden, daß die erste Partition den für die Hash-Tabelle verfügbaren Platz maximal ausnutzt, um das E/A-Verhalten mit zunehmender Hauptspeichergröße zu verbessern. Der Idealfall, daß die innere Relation komplett im Hauptspeicher Platz findet, ergibt sich dann als Spezialfall.

Der Hybrid-Hash-Join verlangt jedoch i.a. eine ungleichmäßige Partitionierung, um den Hauptspeicher für die erste Partition weitgehend nutzen zu können. Zudem ist die maximale Nutzung des Hauptspeichers im Einbenutzerbetrieb zwar sinnvoll, kann allerdings im Mehrbenutzerbetrieb gleichzeitig laufende Transaktionen höherer Priorität stark benachteiligen. Diese Nachteile können durch einen adaptiven Hash-Join umgangen werden, der auf der gleichförmigen Zerlegung in q möglichst gleichgroße Partitionen aufbaut. Ein solcher Ansatz ist der in [PCL93] vorgestellte "Partially Preemptible Hash Join" (PPHJ). Dabei werden für die ersten k der q S-Partitionen die Hash-Tabellen im Hauptspeicher gehalten, für die q-k anderen lediglich ein Ausgabepuffer von einer Seite. Somit läßt sich für die k ersten Partitionen die Join-Berechnung unmittelbar ausführen, und nur für die restlichen Partitionen werden E/A-Vorgänge auf temporäre Dateien nötig. Weiterhin kann die Anzahl der im Hauptspeicher resident gehaltenen S-Partitionen dynamisch variiert werden. So wird deren Anzahl verringert, wenn die Hash-Tabellen wegen ungenauer Schätzungen doch nicht vollständig Platz im Hauptspeicher finden oder wenn aufgrund von Speicheranforderungen durch andere Transaktionen der für die Join-Bearbeitung verfügbare Platz sinkt. Für die betroffenen Partitionen werden dann die Tupel der Hash-Tabelle auf eine temporäre Datrei ausgeschrieben und der Speicherplatz bis auf eine Seite für den Ausgabepuffer freigegeben. Umgekehrt können bei wachsender Hauptspeicherverfügbarkeit während der Partitionierungs-Phase einige bereits auf Platte geschriebene S-Partitionen wieder in den Hauptspeicher gebracht werden, um die weitere Join-Bearbeitung direkt auf ihnen vorzunehmen.

Eine allgemeine Optimierungsmöglichkeit zur Reduzierung von E/A-Vorgängen auf temporären Dateien besteht in der Nutzung von Bitvektoren. Der Bitvektor basiert dabei auf einer auf dem Join-Attribut anzuwendenden Hash-Funktion (z.B. die zur Join-Verarbeitung genutzte Funktion h) und enthält für jede Hash-Klasse ein Bit. Während des Einlesens der S-Relation in der Partitionierungs-Phase wird die Hash-Funktion auf jedes S-Tupel angewendet und das zugehörige Bit gesetzt. Dies kann dann zur Reduzierung der größeren Relation R genutzt werden, da nur noch solche R-Tupel in der weiteren Verarbeitung zu berücksichtigen sind, für die das zugehörige Bit gesetzt ist. Durch diese Verkleinerung der R-Partitionen können oft zahlreiche E/A-Vorgänge eingespart und die Join-Bearbeitung beschleunigt werden. Ein hoher Filtereffekt kann bereits erwartet werden, wenn die Anzahl der Bits doppelt so hoch liegt wie die Tupelanzahl der inneren Relation S [Gra93].

Parallele Realisierungen

Da Hash-Joins zur Berechnung von Equi-Joins eingesetzt werden, eignet sich vor allem eine Parallelisierung mit dynamischer Partitionierung der Eingaberelationen (Kap. 18.3.2). Die unterschiedliche Bearbeitung der inneren und der äußeren Relation im Rahmen der Building- und Probing-Phasen erfordert jedoch, diese beiden Phasen weiterhin nacheinander auszuführen (Abb. 18-5). Es wird also zunächst nur die S-Relation an ihren m Datenknoten parallel eingelesen und unter den p Join-Prozessoren umverteilt. Falls die einzelnen S-Partitionen der Join-Rechner ausreichend klein sind (Basisverfahren), werden sie zudem direkt in Hauptspeicher-Hash-Tabellen gebracht. Danach wird das parallele Lesen und Umverteilen der äußeren Relation gestartet. Im Basisverfahren kann in den Join-Rechnern zudem sofort das Probing vorgenommen werden.

Abb. 18-5: Parallele Hash-Join-Berechnung mit dynamischer Partitionierung der Eingaberelationen

Die Sequentialisierung der beiden Phasen kann als Nachteil gegenüber anderen lokalen Join-Methoden wie Sort-Merge aufgefaßt werden, bei denen beide Relationen parallel verarbeitet werden. Allerdings kann bei Sort-Merge-Joins keine Pipeline-Parallelität zwischen den Daten- und den Join-Knoten genutzt werden. Denn die lokale Join-Bearbeitung ist für Sort-Merge i.a. zu verzögern, bis beide Eingaben vollständig vorliegen, da diese zunächst sortiert werden müssen. Hash-Joins können dagegen Pipeline-Parallelität in beiden Phasen (Building- und Probing-Phasen) einsetzen, was im Rahmen von Mehr-Wege-Joins noch mehr von Vorteil ist (s.u.). Außerdem ist die Sequentialisierung der beiden Phasen Voraussetzung für die Nutzung von Bitvektoren (Hash-Filtern), welche nun auch zu einer signifikanten Reduzierung des Kommunikationsaufwandes zur Datenumverteilung nutzbar sind [SD90]. Hierzu wird an jedem S-Datenknoten vor der Umverteilung ein Bitvektor analog zum zentralen Fall ermittelt, welche dann über eine logische AND-Operation zu einem gemeinsamen Bitvektor für die gesamte Relation verknüpft werden. Dieser Vektor wird dann an die R-Datenknoten übermittelt, womit die Datenumverteilung auf diejenigen R-Tupel beschränkt werden kann, für die das entsprechende Bit gesetzt ist und die daher potentiell am Verbund teilnehmen.

Die dynamische Partitionierung der Eingaberelationen zur Parallelisierung der Join-Bearbeitung ähnelt offenbar der Partitionierung zur Überlaufbehandlung im zentralen Fall. Denn in beiden Fällen wird über eine Hash- oder Bereichspartitionierung eine Zerlegung der beteiligten Relationen in mehrere Partitionen vorgenommen, wobei die Join-Bearbeitung für eine Partition der inneren Relation auf genau eine Partition der äußeren Relation beschränkt wird. Dies gestattet nun eine unmittelbare Parallelisierung. Zudem kann im verteilten Fall aufgrund des (im Mittel um den Faktor p) verringerten Join-Umfangs pro Knoten mit einer höheren Wahrscheinlichkeit das Basisverfahren angewendet werden, bei dem keine Überlaufbehandlung notwendig ist[84].

Wenn die S-Partitionen in den Join-Rechnern jedoch nicht vollständig im Hauptspeicher Platz finden, wird eine zusätzliche dynamische Partitionierung zur Überlaufbehandlung erforderlich. Diese kann prinzipiell vor der Umverteilung an den Datenknoten oder nach der Umverteilung an den Join-Rechnern erfolgen. Wird die Überlaufbehandlung an den Join-Rechnern vorgenommen, ergibt sich folgende Vorgehensweise:

1. Paralleles Lesen und Umverteilen von S
an jedem S-Knoten j (j=1, ..., m) führe parallel durch:
lies lokale S-Partition Sj und sende jedes S-Tupel an zuständigen Join-Rechner (Anwendung der Verteilungsfunktion f auf dem Join-Attribut)

2. Paralleles Partitionieren der S-Partitionen
in jedem Join-Knoten k führe parallel durch (k=1...p)
- Sk' := Sjk (j=1..m) (Menge der von Join-Knoten k empfangenen S-Tupel)
- Partitionierung in q Sub-Partitionen zur Überlaufbehandlung Sk1' bis Skq'
durch Anwendung einer Partitionierungsfunktion g auf dem Join-Attribut

3. Paralles Lesen und Umverteilen von R
an jedem R-Knoten i (i=1..n) führe parallel durch:
lies lokale R-Partition Ri und sende jedes R-Tupel an zuständigen Join-Rechner (Verteilungsfunktion f)

4. Paralleles Partitionieren der R-Partitionen
in jedem Join-Knoten k führe parallel durch (k=1...p)
- Rk' := Rik (i=1..n) (Menge der von Join-Knoten k empfangenen R-Tupel)
- Partitionierung in q Sub-Partitionen zur Überlaufbehandlung Rk1' bis Rkq'
durch Anwendung einer Partitionierungsfunktion g auf dem Join-Attribut
Nach diesem Schritt ergibt sich die in Abb. 18-6 gezeigte Situation.

5. Parallele Join-Berechnung
in jedem Join-Knoten k führe parallel durch (k=1...p):
für l = 1, 2 bis q führe nacheinander aus:
berechne Skl' Rkl'
schicke Ergebnis an Koordinator

Abb. 18-6: Zweistufige dynamische Partitionierung der Eingaberelationen (Überlaufbehandlung an den Join-Rechnern)

Der gezeigte Algorithmus nutzt Datenparallelität an den Datenknoten zum paralleles Einlesen (aufgrund der statischen Datenpartitionierung) sowie an den Join-Knoten zum parallelen Umverteilen, Partitionieren zur Überlaufbehandlung, sowie zur Join-Berechnung (dynamische Datenpartitionierung). Umverteilung und Überlaufbehandlung der beiden Eingaberelationen sind hier prinzipiell auch parallel möglich (Schritte 1 und 2 parallel zu den Schritten 3 und 4). Pipeline-Parallelität kann zwischen den Daten- und Join-Knoten genutzt werden, wobei auch das Ausschreiben der temporären Partitionen Skl' und Rkl' an den Join-Rechnern über Ausgabepuffer weitgehend asynchron realisierbar ist.

Die Überlaufbehandlung an den Join-Rechnern bringt den Vorteil mit sich, daß dort praktisch jeder lokale Ansatz gewählt werden kann. So entspricht der gezeigte Algorithmus der parallelen Version eines GRACE-Hash-Joins. Die Realisierung eines parallelen Hybrid-Hash-Joins ist analog möglich. Hierzu wird in jedem Join-Rechner k bei der Überlaufbehandlung die jeweils erste S-Partition Sk1' im Hauptspeicher gehalten, um E/A-Vorgänge einzusparen und mit den zugehörigen R-Tupeln die Join-Berechnung sofort vornehmen zu können (Voraussetzung hierfür ist jedoch wieder die Sequentialisierung von S- und R-Bearbeitung). In ähnlicher Weise können auch adaptive Hash-Verfahren in den Join-Rechnern angewendet werden.

Alternativ zu der beschriebenen Vorgehensweise kann eine Überlaufbehandlung bereits an den Datenknoten erfolgen (s. Übungsaufgaben). Ein Vorteil dabei ist die Vermeidung von E/A-Vorgängen an den Join-Rechnern, so daß selbst Rechner ohne Platten (diskless nodes) zur Join-Berechnung nutzbar werden.

Auch für die parallele Hash-Join-Berechnung kann die Anzahl der Join-Prozessoren sowie ihre Auswahl wieder dynamisch - in Abhängigkeit der aktuellen Systemauslastung - festgelegt werden. Die Untersuchung [RM95] zeigte, daß hierfür vor allem die aktuelle Hauptspeicherverfügbarkeit der einzelnen Rechner berücksichtigt werden sollte, um eine Zuordnung zu finden, die mit möglichst wenigen E/A-Vorgängen zur Überlaufbehandlung auskommt. Dazu ist es bei hoher Hauptspeicherauslastung empfehlenswert, eine höhere Anzahl von Join-Prozessoren zu wählen, um den Speicherbedarf pro Knoten zu reduzieren. Dies ist die entgegengesetzte Vorgehensweise wie zur Behandlung von CPU-Engpässen, bei denen sich zur Reduzierung des Kommunikations-Overheads eine Reduzierung des Paralleltätsgrades empfiehlt.

[84] Die Speicherung der inneren Relation im Rahmen von Hash-Tabellen entspricht einer weiteren Partitionierung, die den Suchraum zur Bestimmung der Verbundpartner eingrenzt.