View a markdown version of this page

Dynamische Partitionen beschneiden - AWS Prescriptive Guidance

Die vorliegende Übersetzung wurde maschinell erstellt. Im Falle eines Konflikts oder eines Widerspruchs zwischen dieser übersetzten Fassung und der englischen Fassung (einschließlich infolge von Verzögerungen bei der Übersetzung) ist die englische Fassung maßgeblich.

Dynamische Partitionen beschneiden

Spark3.0 und höher beinhalten Dynamic Partition Pruning (DPP). Beim dynamischen Partitionsbereinigen handelt es sich um eine OptimierungstechnikSpark, die verhindert, dass beim Lesen von Daten nicht benötigte Partitionen gescannt werden. Im Folgenden sind einige wichtige Dinge aufgeführt, die Sie über DPP wissen sollten:

  • Es untersucht die in den Abfragefiltern und Prädikaten angeforderten Partitionswerte und bestimmt, welche Partitionen erforderlich sind, um die Abfrage zu erfüllen. Alle Partitionen, die als unnötig erachtet werden, werden automatisch und transparent gelöscht.

  • DPP reduziert die Verarbeitungszeit und die Ressourcennutzung, indem Partitionen übersprungen werden, die keine entsprechenden Daten enthalten. Dies hilft, sich nur auf relevante Partitionen Spark zu konzentrieren.

  • Es funktioniert sowohl mit statischen Partitionen als auch mit dynamisch generierten Partitionen, die durch Einfügungen oder inkrementelles Laden hinzugefügt werden. Sparkerkennt neue Partitionen und kann weiterhin dynamisches Bereinigen anwenden.

  • DPP ist für Entwickler völlig transparent oder unsichtbar. Für die Aktivierung von DPP ist keine spezielle Codierung erforderlich. Es erfolgt automatisch hinter den Kulissen als Optimierung bei der Generierung von Abfrageplänen.

Im Folgenden finden Sie einige bewährte Methoden, um sicherzustellen, dass DPP effizient funktioniert:

  • Verwenden Sie Predicate Pushdown, indem Sie Filter schon früh in Ihren Spark Datenrahmenoperationen anwenden. Dies hilft dabeiSpark, Partitionen frühzeitig zu entfernen, indem die Partitionsmetadaten verwendet werden.

  • Sammeln Sie Statistiken zu Ihren Daten, indem Sie sie ANALYZE TABLE häufig ausführen. Dadurch werden Statistiken auf Spaltenebene reduziert, mit deren Hilfe Spark genauer bestimmt werden kann, welche Partitionen ignoriert werden können.

  • Vermeiden Sie eine Überpartitionierung Ihrer Daten. Zu viele Partitionen können den Treiberknoten beim Sammeln von Statistiken überlasten. Streben Sie für jede große Tabelle 10—100 Partitionen an.

  • Partitionieren Sie Datenframes neu, bevor Sie sie verknüpfen. Dadurch werden Zufallsverknüpfungen verhindert, bei denen alle Daten verschoben werden müssen, und die Menge der gelesenen Daten wird weiter optimiert.

  • Verwenden Sie konsistente Partitionsspaltentypen und Benennungen für verschiedene Tabellen, die verknüpft werden. Dies hilft, Partitionen für die Join-Optimierung Spark besser aufeinander abzustimmen.

  • Testen Sie Abfragen mitEXPLAIN, um sicherzustellen, dass DPP angewendet wird, und überprüfen Sie, ob zusätzliche Anpassungen erforderlich sind.

In einem Sternschema werden Tabellen in zwei Haupttypen unterteilt: Faktentabellen und Dimensionstabellen. Dimensionstabellen sind in der Regel viel kleiner als Faktentabellen. Wenn eine Faktentabelle mit einer Dimensionstabelle verknüpft wird, optimiert DPP den Abfrageplan. Es erstellt eine Unterabfrage aus allen Filtern, die auf die Dimensionstabelle angewendet werden. Es sendet diese Unterabfrage und erstellt daraus eine Hashtabelle. Anschließend wendet es die Hashtabelle auf die Scan-Phase der Faktentabelle an, bevor die Daten der Faktentabelle gelesen werden. Auf diese Weise kann DPP die Datenmenge reduzieren, die aus der größeren Faktentabelle gelesen werden muss.

Die folgende Beispielabfrage zeigt DPP in Aktion. Die Abfrage ruft die Anzahl der Bestellungen aus dem Land (Indien) ab und beinhaltet eine innere Verknüpfung zwischen einer Faktentabelle (fact_orders) und einer Dimensionstabelle (nation). Die fact_orders Tabelle ist nach der Spalte partitioniert. o_nationkey

- "select n.n_name as country, count(1) as no_of_orders from fact_orders o join nation n on o.o_nationkey = n.n_nationkey where n.n_name = 'INDIA' group by n.n_name"

Im EXPLAIN Plan werden die folgenden Schritte verwendet:

  1. Scannen Sie die Tabelle mit den kleineren Dimensionen (nation) und filtern Sie nach Spaltenn_name = 'INDIA'.

  2. Senden Sie die Ergebnisse des vorherigen Schritts.

  3. Erstellen Sie eine Unterabfrage, die nach den Ergebnissen des ersten Schritts filtert.

  4. Verschieben Sie sie als eine, PartitionFilter sodass sie nur die Partitionen der Faktentabellen scannt, die tatsächlich benötigt werden, und nicht einen vollständigen Tabellenscan.

Im Folgenden finden EXPLAIN Sie den Plan für diese DPP-optimized Abfrage.

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=true
+- == Final Plan ==
   *(4) HashAggregate(keys=[], functions=[count(1)], output=[count#208L])
   +- ShuffleQueryStage 3
      +- Exchange SinglePartition, ENSURE_REQUIREMENTS, [id=#353]
         +- *(3) HashAggregate(keys=[], functions=[partial_count(1)], output=[count#212L])
            +- *(3) HashAggregate(keys=[n_name#31], functions=[], output=[])
               +- ShuffleQueryStage 1
                  +- Exchange hashpartitioning(n_name#31, 36), ENSURE_REQUIREMENTS, [id=#315]
                     +- *(2) HashAggregate(keys=[n_name#31], functions=[], output=[n_name#31])
                        +- *(2) Project [n_name#31]
                           +- *(2) BroadcastHashJoin [cast(o_nationkey#145 as bigint)], [n_nationkey#32L], Inner, BuildRight, false
                              :- *(2) ColumnarToRow
                              :  +- FileScan parquet [o_nationkey#145] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex[s3://aws-spark-tuning/fact_orders], PartitionFilters: [isnotnull(o_nationkey#145), dynamicpruningexpression(cast(o_nationkey#145 as bigint) IN dynamicp..., PushedFilters: [], ReadSchema: struct<>
                              :        +- SubqueryBroadcast dynamicpruning#210, 0, [n_nationkey#32L], [id=#200]
                              :           +- OutputAdapter [n_name#31, n_nationkey#32L]
                              :              +- AdaptiveSparkPlan isFinalPlan=true
                              :                 +- BroadcastQueryStage 2
                              :                    +- ReusedExchange [n_name#31, n_nationkey#32L], BroadcastExchange HashedRelationBroadcastMode(List(input[1, bigint, false]),false), [id=#233]
                              +- BroadcastQueryStage 0
                                 +- BroadcastExchange HashedRelationBroadcastMode(List(input[1, bigint, false]),false), [id=#233]
                                    +- *(1) Filter ((isnotnull(n_name#31) AND (n_name#31 = INDIA)) AND isnotnull(n_nationkey#32L))
                                       +- FileScan json [n_name#31,n_nationkey#32L] Batched: false, DataFilters: [isnotnull(n_name#31), (n_name#31 = INDIA), isnotnull(n_nationkey#32L)], Format: JSON, Location: InMemoryFileIndex[s3://aws-spark-tuning/input/demo/json/nation], PartitionFilters: [], PushedFilters: [IsNotNull(n_name), EqualTo(n_name,INDIA), IsNotNull(n_nationkey)], ReadSchema: struct<n_name:string,n_nationkey:bigint>

Obwohl der o_nationkey Spalte kein direkter Filter hinzugefügt wurde, werden aufgrund der DPP-Funktion Spark automatisch nur die Partitionen gescannt, die benötigt werden, und nicht die gesamte Tabelle.