Udostępnij przez

Optymalizacja przetwarzania danych dla platformy Apache Spark

W tym artykule omówiono sposób optymalizacji konfiguracji klastra Apache Spark w celu uzyskania najlepszej wydajności w usłudze Azure HDInsight.

Przegląd

Jeśli masz powolne zadania w sprzężeniu lub przetwarzaniu Shuffle, przyczyną jest prawdopodobnie niesymetryczność danych. Skew danych to asymetria w danych zadań. Na przykład zadanie mapy może potrwać 20 sekund. Jednak uruchomienie zadania, w którym dane są połączone lub przetasowane, może zająć godziny. Aby naprawić niesymetryczność danych, należy skonsolić cały klucz lub użyć izolowanej soli tylko dla niektórych podzestawów kluczy. Jeśli używasz izolowanej soli, należy dodatkowo filtrować, aby odizolować podzbiór kluczy solnych w sprzężeniach mapy. Inną opcją jest wprowadzenie kolumny zasobnika i wstępne agregowanie w zasobnikach.

Innym czynnikiem powodującym powolne sprzężenia może być typ sprzężenia. Domyślnie platforma Spark używa SortMerge typu sprzężenia. Ten typ sprzężenia najlepiej nadaje się do obsługi dużych zestawów danych. Ale jest inaczej kosztowny obliczeniowo, ponieważ musi najpierw posortować lewe i prawe strony danych przed ich scaleniem.

Sprzężenia Broadcast najlepiej nadaje się do mniejszych zestawów danych lub gdy jedna strona sprzężenia jest znacznie mniejsza niż po drugiej stronie. Ten typ łączenia emituje jedną stronę do wszystkich wykonawców, dlatego ogólnie wymaga więcej pamięci na transmisje.

Typ sprzężenia można zmienić w konfiguracji, ustawiając spark.sql.autoBroadcastJoinThreshold, lub możesz ustawić wskazówkę sprzężenia, korzystając z interfejsów API ramki danych (dataframe.join(broadcast(df2))).

// Option 1
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 1*1024*1024*1024)

// Option 2
val df1 = spark.table("FactTableA")
val df2 = spark.table("dimMP")
df1.join(broadcast(df2), Seq("PK")).
    createOrReplaceTempView("V_JOIN")

sql("SELECT col1, col2 FROM V_JOIN")

Jeśli używasz tabel podzielonych na segmenty, masz trzeci typ sprzężenia, sprzężenie Merge. Prawidłowo przygotowany i wstępnie posortowany zestaw danych pominie kosztowną fazę sortowania w ramach sprzężenia SortMerge.

Kolejność sprzężeń ma znaczenie, szczególnie w bardziej złożonych zapytaniach. Zacznij od najbardziej selektywnych sprzężeń. Ponadto, jeśli to możliwe, przenoś sprzężenia, które zwiększają liczbę wierszy po agregacjach.

Aby zarządzać równoległością sprzężeń kartezjańskich, możesz dodać zagnieżdżone struktury, okna i być może pominąć jeden lub więcej kroków w zadaniu Spark.

Optymalizowanie wykonywania zadań

  • Buforuj w razie potrzeby, na przykład jeśli używasz danych dwa razy, a następnie buforuj je.
  • Rozgłasz zmienne do wszystkich funkcji wykonawczych. Zmienne są serializowane tylko raz, co powoduje szybsze wyszukiwanie.
  • Użyj puli wątków w sterowniku, co przyspiesza działanie wielu zadań.

Regularnie monitoruj uruchomione zadania pod kątem problemów z wydajnością. Jeśli potrzebujesz więcej szczegółowych informacji na temat niektórych problemów, rozważ jedną z następujących narzędzi profilowania wydajności:

Kluczem do wydajności zapytań platformy Spark 2.x jest silnik Tungsten, który opiera się na generowaniu kodu na pełnym etapie. W niektórych przypadkach generowanie kodu pełnoetapowego może być wyłączone. Na przykład, jeśli użyjesz w wyrażeniu agregacyjnym typu niezmienialnego (string), zamiast tego pojawi się HashAggregate. Aby na przykład uzyskać lepszą wydajność, spróbuj wykonać następujące czynności, a następnie ponownie włączyć generowanie kodu:

MAX(AMOUNT) -> MAX(cast(AMOUNT as DOUBLE))

Następne kroki

  • Last updated on