Załóżmy, że masz sytuację, w której jeden zestaw danych jest bardzo mały, a drugi dość duży i chcesz wykonać operację join między tymi dwoma. W takim przypadku powinniśmy wybrać opcję broadcast join, aby mały zestaw danych pasował do zmiennej broadcast. Składnia do użycia zmiennej broadcast to df1.Dołącz(broadcast(DF2)). Tutaj mamy drugą ramkę danych, która jest bardzo mała i zachowujemy tę ramkę danych jako zmienną transmisji.
Code:
broadcast (tablica(1, 2, 3))
broadcastVar.wartość
res0: Array = Array(1, 2, 3)
val accum = sc.longAccumulator („mój akumulator”)
sc.paralelize (tablica(1, 2, 3, 4)).foreach (x = > accum.dodaj (x))
wartość
res2: Long = 10
Cache i Persist
- Spark udostępnia własne mechanizmy buforowania, takie jak persist() i cache().
- cache() i persist() przechowują zestaw danych w pamięci.
- gdy masz mały zestaw danych, który musi być użyty wiele razy w programie, buforujemy ten zestaw danych.
- Cache() – zawsze w pamięci
- Persist() – pamięć i dyski
Spark zapewnia własny mechanizm buforowania, taki jak Persist i buforowanie. Mechanizmy Persist i Cache przechowują zestaw danych w pamięci, gdy jest to wymagane, gdzie masz mały zestaw danych i ten zestaw danych jest używany wiele razy w programie. Jeśli zastosujemy RDD.Cache() zawsze będzie przechowywać dane w pamięci, a jeśli zastosujemy RDD.Persist() wtedy część danych może być zapisana do pamięci, część może być zapisana na dysku.
5. Operacja ByKey
- tasowania to ciężka operacja, która zużywa dużo pamięci.
- podczas kodowania w Spark Użytkownik powinien zawsze unikać operacji tasowania.
- wysokie tasowanie może spowodować błąd poza pamięcią; aby uniknąć takiego błędu, użytkownik może zwiększyć poziom równoległości.
- użyj reduceByKey zamiast groupByKey.
- poprawne dzielenie danych.
jak wiemy podczas naszej transformacji Spark mamy wiele operacji ByKey. Operacje ByKey generują wiele przetasowań. Przetasowania są ciężkie, ponieważ zużywają dużo pamięci. Podczas kodowania w Spark Użytkownik powinien zawsze unikać operacji losowania, ponieważ spowoduje to pogorszenie wydajności. Jeśli istnieje wysokie tasowanie, użytkownik może usunąć błąd z pamięci. W tym przypadku, aby uniknąć tego błędu, Użytkownik powinien zwiększyć poziom równoległości. Zamiast groupBy, Użytkownik powinien wybrać reduceByKey, ponieważ groupByKey tworzy wiele tasowań, które hamują wydajność, podczas gdy reduceByKey nie tasuje danych tak bardzo. Dlatego reduceByKey jest szybszy niż groupByKey. Za każdym razem, gdy używana jest jakakolwiek operacja ByKey, Użytkownik powinien poprawnie partycjonować dane.
6. Wybór formatu pliku
- Spark obsługuje wiele formatów, takich jak CSV, JSON, XML, PARQUET, ORC, AVRO itp.
- zadania Spark można zoptymalizować, wybierając plik parkietowy z snappy compression, co zapewnia wysoką wydajność i najlepszą analizę.
- plik Parquet jest natywny dla Spark, który przenosi metadane wraz ze stopką.
Spark ma wiele formatów plików, takich jak CSV, JSON, XML, parkiet, ORC, AVRO i inne. Zadanie iskry można zoptymalizować, wybierając plik parkietowy z błyskawiczną kompresją. Plik parkiet jest natywny dla Spark, które przenoszą metadane wraz ze stopką, jak wiemy, plik parkiet jest natywny dla spark, który jest w formacie binarnym, a wraz z danymi, które również przenoszą stopkę, a także przenoszą metadane i stopkę, więc za każdym razem, gdy tworzysz dowolny plik parkiet, zobaczysz .plik metadanych w tym samym katalogu wraz z plikiem danych.
Kod:
Czytaj.json („examples / src/main/resources / people.json”)
osób.pisz.parkiet („lud.parkiet”)
val parquetFileDF = spark.Czytaj.parkiet („lud.parkiet”)
val usersDF = spark.Czytaj.format („avro”).load („examples / src / main / resources / users.avro”)
użytkowników.select(„name”, „favorite_color”).pisz.format („avro”).save („namesAndFavColors.avro”)
Tuning Garbage Collection
- JVM garbage collection może być problemem, gdy masz duży zbiór nieużywanych obiektów.
- pierwszym krokiem w GC tuning jest zbieranie statystyk poprzez wybranie opcji-verbose podczas przesyłania zadań spark.
- w idealnej sytuacji staramy się utrzymać koszty ogólne GC < 10% pamięci sterty.
jak wiemy, nasze zadanie Spark działa na platformie JVM, więc JVM garbage collection może być problematyczne, gdy masz dużą kolekcję nieużywanego obiektu, więc pierwszym krokiem w strojeniu garbage collection jest zbieranie statyki, wybierając opcję w Twoim Spark submit verbose. Ogólnie rzecz biorąc, w idealnej sytuacji powinniśmy zachować pamięć do zbierania śmieci mniej niż 10% pamięci sterty.
8. Poziom równoległości
- równoległość odgrywa bardzo ważną rolę podczas dostrajania zadań iskry.
- każda partycja ~ task wymaga jednego rdzenia do przetwarzania.
- istnieją dwa sposoby utrzymania równoległości:
- Repartition: daje równą liczbę partycji z wysokim tasowaniem
- Coalesce: ogólnie zmniejsza liczbę partycji z mniejszym tasowaniem.
w każdym rozproszonym środowisku równoległość odgrywa bardzo ważną rolę podczas dostrajania zadania Spark. Ilekroć przesyłane jest zadanie Spark, tworzy ono biurko, które będzie zawierało etapy, a zadania zależą od partycji, więc każda partycja lub zadanie wymaga pojedynczego rdzenia systemu do przetwarzania. Istnieją dwa sposoby utrzymania równoległości-podział i łączenie. Za każdym razem, gdy zastosujesz metodę podziału, daje ona taką samą liczbę partycji, ale będzie ona dużo tasować, więc nie jest wskazane, aby przejść do podziału, gdy chcesz usunąć wszystkie dane. Coalesce na ogół zmniejsza liczbę partycji i powoduje mniejsze tasowanie danych.
te czynniki dla optymalizacji iskry, jeśli są prawidłowo stosowane, mogą –
