Adaptacyjne wykonywanie zapytań

Wykonywanie zapytań adaptacyjnych (AQE) to ponowna optymalizacja zapytań wykonywana podczas wykonywania zapytania.

Przed uruchomieniem zapytania optymalizator wybiera plan wykonania zapytania na podstawie szacunków rozmiaru danych. Te szacunki mogą być błędne, jeśli statystyki tabelowe są nieobecne lub przestarzałe, jeśli dane są zniekształcone lub w późniejszych etapach skomplikowanego zapytania. Plan oparty na błędnych szacunkach może stosować strategię powolnego łączenia, tworzyć zbyt wiele małych zadań po tasowaniu lub kilka przerośniętych zadań, które opóźniają cały etap. AQE rozwiązuje te problemy, ponownie optymalizując plan podczas wykonywania zapytania, w oparciu o rzeczywisty rozmiar danych.

Na końcu każdego tasowania lub wymiany broadcast Azure Databricks ma dokładne statystyki dotyczące dotychczas wygenerowanych danych. AQE wykorzystuje te statystyki do wyboru lepszej strategii fizycznej, optymalnej liczby i rozmiaru partycji po tasowaniu lub stosowania optymalizacji, które w innym przypadku wymagałyby podpowiedzi zapytań, takich jak obsługa skośnych złączeń.

Usługa AQE jest domyślnie włączona. Nie musisz zmieniać zapytań, żeby z tego skorzystać. Sprawdź, czy AQE zmieniło plan zapytania podczas rozwiązywania problemów z wolnym zapytaniem. Dostosowuj ustawienia AQE tylko wtedy, gdy domyślne nie pasują do twojego obciążenia roboczego. Zobacz Plany zapytań i konfigurację.

Możliwości

AQE posiada cztery główne cechy:

• Dynamicznie zmienia połączenie z sortowaniem i scalaniem na połączenie haszowe z rozgłaszaniem.
• Dynamicznie konsoliduje partycje (łączenie małych partycji w rozsądnie duże partycje) po wymianie typu shuffle. Bardzo małe zadania mają gorszą przepływność operacji we/wy i mają tendencję do większego obciążenia związanego z planowaniem i konfiguracją zadań. Łączenie małych zadań pozwala zaoszczędzić zasoby i zwiększyć przepływność klastra.
• Dynamicznie obsługuje skos w sort merge join i shuffle hash join, poprzez podzielenie (i replikowanie w razie potrzeby) zadań o nierównomiernym obciążeniu na zadania o mniej więcej równomiernym rozmiarze.
• Dynamicznie wykrywa i propaguje puste relacje, które są częściami zapytania, które nie zwracają żadnych wierszy.

Obsługiwane zapytania

AQE dotyczy zapytań spełniających oba te warunki:

• Zapytanie nie jest zapytaniem streamingowym.
• Zapytanie zawiera co najmniej jeden exchange (zwykle z operacji join, agregacji lub funkcji okna), co najmniej jedno podzapytanie lub oba.

AQE ponownie optymalizuje każde obsługiwane zapytanie, ale wynikiem może być ten sam plan, który optymalizator skompilował przed uruchomieniem zapytania. Aby sprawdzić, czy AQE zmieniło plan zapytania, zobacz Plan zapytania.

Plany zapytań

Zapytania zastosowane w usłudze AQE zawierają co najmniej jeden węzeł AdaptiveSparkPlan, zwykle jako węzeł główny każdego zapytania głównego lub zapytania podrzędnego. Każdy AdaptiveSparkPlan węzeł ma isFinalPlan flagę. Flaga ma wartość false przed i w trakcie wykonywania zapytania i po zakończeniu zapytania zmienia się na true.

Aby zobaczyć plan zapytania, użyj interfejsu Spark, DataFrame.explain(), lub SQL EXPLAIN. Każdy z nich pokazuje inne informacje o tym, jak AQE zmienia plan.

Interfejs użytkownika platformy Spark

Diagram planu zapytania w interfejsie Spark pokazuje najnowszy plan i aktualizuje się w miarę wykonywania zapytania. Węzły, które już działały i mają metryki, nie zmieniają się. Węzły, które jeszcze nie zostały uruchomione, mogą się zmienić, gdy AQE ponownie optymalizuje plan.

Ten przykład przedstawia diagram planu zapytań:

diagram planu zapytania

DataFrame.explain()

DataFrame.explain() pokazuje dwa plany pod każdym węzłem AdaptiveSparkPlan: plan początkowy, zanim AQE zastosuje jakiekolwiek optymalizacje, oraz obecny plan. Obecny plan zmienia się w trakcie działania zapytania i staje się planem końcowym po zakończeniu zapytania.

Dane wyjściowe zawierają statystyki danych dla każdego etapu przetasowania i broadcast. Statystyki zależą od tego, czy etap został uruchomiony:

• Przed i w trakcie wykonania etapu isRuntime i false, a statystyki to szacunki czasu kompilacji. Na przykład Statistics(sizeInBytes=1024.0 KiB, rowCount=4, isRuntime=false).
• Po uruchomieniu etapu isRuntime ma stan true, a statystyki są pomiarami czasu wykonywania. Na przykład Statistics(sizeInBytes=658.1 KiB, rowCount=2.81E+4, isRuntime=true).

Te zrzuty ekranu pokazują DataFrame.explain() dane wyjściowe przed, w trakcie i po wykonaniu:

• Przed wykonaniem

  

• Podczas wykonywania

  

• Po wykonaniu

  

SQL EXPLAIN

SQL EXPLAIN nie uruchamia zapytania, więc AQE nie może go ponownie zoptymalizować. Wynik pokazuje jedynie plan początkowy, który może różnić się od planu wykonywanego przez AQE.

Ten przykład pokazuje SQL EXPLAIN wyjście:

Zidentyfikuj zastosowane optymalizacje w planie zapytań

Gdy AQE stosuje optymalizację, plan zapytań się zmienia. Aby określić, które optymalizacje przyniosły efekt, porównaj obecny lub ostateczny plan z planem początkowym. Każda optymalizacja zmienia plan w określony sposób:

• Zmiana strategii łączenia. Obecny lub ostateczny plan ma inny fizyczny węzeł łączenia niż pierwotny plan.

  

• Scalanie partycji. Plan zawiera CustomShuffleReader węzeł z tą Coalesced właściwością.

  

  

• Obsługa skośnych złączy. Plan zawiera węzeł SortMergeJoin z polem isSkew ustawionym na true.

  

  

• Propagacja pustej relacji. LocalTableScan Węzeł z pustym polem relacji zastępuje część lub całość planu.

  

  

Konfiguracja

Te właściwości konfiguracyjne Apache Spark kontrolują AQE. Większość obciążeń nie musi zmieniać domyślnych ustawień.

Włącz AQE i dostosuj poszczególne optymalizacje

Wykorzystaj te właściwości, aby włączyć lub wyłączyć AQE oraz dostroić automatycznie zoptymalizowane shuffle, przełączanie złączenia rozgłoszeniowego oraz propagację pustych relacji:

Własność	Typ	Default	Opis
spark.databricks.optimizer.adaptive.enabled	Boolean	true	Włącza lub wyłącza adaptacyjne wykonywanie zapytań.
spark.sql.shuffle.partitions	Integer	200	Domyślna liczba partycji do użycia podczas mieszania danych dla sprzężeń lub agregacji. Ustaw ją na auto, aby włączyć automatyczne zoptymalizowane tasowanie, które ustawia tę liczbę na podstawie planu zapytania i rozmiaru danych wejściowych zapytania. W przypadku Structured Streaming nie możesz zmieniać tej wartości między restartami zapytań z tej samej lokalizacji punktu kontrolnego.
spark.databricks.adaptive.autoBroadcastJoinThreshold	Byte String	30MB	Próg rozmiaru umożliwiający przełączenie złączenia sortująco-scalającego na broadcast hash join w czasie działania.
spark.databricks.adaptive.emptyRelationPropagation.enabled	Boolean	true	Włącza lub wyłącza propagację dynamicznej pustej relacji.

Dynamiczne scalanie partycji

Własność	Typ	Default	Opis
spark.sql.adaptive.coalescePartitions.enabled	Boolean	true	Włącza lub wyłącza łączenie partycji.
spark.sql.adaptive.advisoryPartitionSizeInBytes	Byte String	64MB	Rozmiar docelowy po połączeniu. Scalone partycje są zbliżone do tego rozmiaru, ale nie większe od niego.
spark.sql.adaptive.coalescePartitions.minPartitionSize	Byte String	1MB	Minimalny rozmiar partycji po scaleniu. Scalone partycje nie są mniejsze niż ten rozmiar.
spark.sql.adaptive.coalescePartitions.minPartitionNum	Integer	Dwukrotność liczby rdzeni klastrowych	Minimalna liczba partycji po scaleniu. Nie ustawiaj tej właściwości, ponieważ jej ustawienie nadpisuje spark.sql.adaptive.coalescePartitions.minPartitionSize.

Dynamiczna obsługa łączenia skośnego

Własność	Typ	Default	Opis
spark.sql.adaptive.skewJoin.enabled	Boolean	true	Włącza lub wyłącza obsługę skew join.
spark.sql.adaptive.skewJoin.skewedPartitionFactor	Integer	5	Mnożnik mediany rozmiaru partycji, który partcja musi przekroczyć, aby uznać ją za skośną.
spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes	Byte String	256MB	Minimalny rozmiar, który partycja musi przekroczyć, aby uznać ją za nierównomiernie rozłożoną.

AQE traktuje podział jako skośny tylko wtedy, gdy oba te warunki są prawdziwe: (partition size > skewedPartitionFactor * median partition size) oraz (partition size > skewedPartitionThresholdInBytes).

Często zadawane pytania

Dlaczego AQE nie emitowała małej tabeli sprzężenia?

Jeśli tabela jest mniejsza niż spark.databricks.adaptive.autoBroadcastJoinThreshold, ale AQE jej nie emituje, sprawdź następujące przyczyny:

• Typ łączenia nie obsługuje broadcast. Niektóre typy połączeń nie mogą rozgłaszać jednej strony. Na przykład AQE nie może rozsyłać lewej strony LEFT OUTER JOIN sprzężenia.
• Tabela zawiera wiele pustych partycji. Większość zadań w operacji sort merge join kończy się szybko albo obsługa skew join optymalizuje złączenie. AQE nie zmienia złączenia sortująco-scalającego na broadcast hash join, jeśli procent niepustych partycji jest niższy niż spark.sql.adaptive.nonEmptyPartitionRatioForBroadcastJoin.

Czy powinienem nadal stosować wskazówkę broadcast join przy włączonym AQE?

Tak, jeśli dobrze znasz swoje zapytanie. Połączenie broadcastowe, które optymalizator planuje przed uruchomieniem zapytania, zwykle działa lepiej niż to, które AQE planuje w czasie działania. AQE może przełączyć się na broadcast join dopiero po przetasowaniu obu stron join, wtedy pozna rzeczywiste rozmiary tabel.

AQE traktuje podpowiedzi zapytań tak samo jak optymalizacja statyczna i nadal stosuje optymalizacje dynamiczne, na które podpowiedzi nie wpływają.

Czy powinienem użyć wskazówki dotyczącej skew join czy obsługi skew join AQE?

Użyj obsługi skośnych złączeń AQE. Jest to automatyczne i zwykle działa lepiej niż wskazówka skew join.

Dlaczego AQE nie zmieniło kolejności moich złączeń?

AQE nie zmienia kolejności połączeń w czasie działania. Aby poprawić kolejność złączeń, zbieraj statystyki tabelowe, aby optymalizator oparty na kosztach mógł wybrać lepszy plan. Zobacz optymalizator oparty na kosztach.

Dlaczego usługa AQE nie wykryła niesymetryczności danych?

AQE traktuje podział jako przekoszony tylko wtedy, gdy spełnione są oba te warunki:

• Partycja jest większa niż spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes (domyślnie 256MB).
• Partycja jest większa niż spark.sql.adaptive.skewJoin.skewedPartitionFactor razy mediana rozmiaru partycji (domyślnie 5).

AQE obsługuje też skos tylko w połączeniach typu shuffle, które są połączeniami sortowania przez scalanie i połączeniami haszującymi typu shuffle. Nigdy nie optymalizuje nierównomierności w złączeniach rozgłoszeniowych. Dla złączeń opartych na tasowaniu typ łączenia określa, którą stronę AQE może optymalizować:

Typ sprzężenia	Skos zoptymalizowany po lewej stronie	Skos zoptymalizowany po prawej stronie
INNER	Yes	Yes
CROSS	Yes	Yes
LEFT OUTER	Yes	Nie.
RIGHT OUTER	Nie.	Yes
LEFT SEMI	Yes	Nie.
LEFT ANTI	Yes	Nie.
FULL OUTER	Nie.	Nie.

Dziedzictwo

Spark 1.6 wprowadził funkcję zwaną adaptive execution, ale AQE w Spark 3.0 to inna funkcja. Adaptacyjne wykonywanie Spark 1.6 łączy partycje tylko dynamicznie. AQE to framework, który planuje i przeplanowuje zapytania na podstawie statystyk z czasu wykonywania, wspierając wszystkie optymalizacje opisane na tej stronie.