Big Data w New Space – technologie BD w branży kosmicznej

Big Data w New Space – technologie BD w branży kosmicznej

XXI wiek to okres głębokiej rewolucji w kwestiach kosmicznych. Najpierw kosmos został kompletnie zdeprecjonowany (łącznie z rozważaniami nad likwidacją NASA za Barracka Obamy). Następnie powstały bezprecedensowe próby rozwinięcia podboju kosmosu przez… sektor prywatny, z Elonem Muskiem na czele. Potem do gry weszli Chińczycy, którzy postawili Stanom Zjednoczonym potężne wyzwanie i… zaczęło się. Cała ta wielka przygoda nie mogła oczywiście odbyć się bez nowoczesnych technologii przetwarzania danych. Tak Big Data weszła do New Space.

New Space

Sektor prywatny pokazał, że do kosmosu można podchodzić w zupełnie nowy sposób“Po rynkowemu” – z konkurencją, obniżając ceny, grając jakością, wskazując zupełnie nowe pola do rozwoju, a nade wszystko – robiąc na tym solidny biznes.

Odpowiedzią na rozwój sytuacji po stronie Chin oraz rodzących się nowych możliwości były śmiałe pomysły rządu Amerykańskiego – program powrotu człowieka na księżyc “Artemis” (wraz z Artemis accords) oraz utworzenie Space Force (sił kosmicznych).

Powstała całkowicie nowa domena – ochrzczona jako New SpaceWraz z “Baronami kosmosu” (wielkimi przedsiębiorcami wykładającymi swoje pieniądze na rozwój sektora kosmicznego), rywalizacją między mocarstwami i… coraz szybszym postępem technologicznym, który wykorzystywany jest obficie w życiu codziennym milionów osób.

A teraz pytanie retoryczne: czy mogą tak olbrzymie posunięcia technologiczne obyć się bez Big Data? Odpowiedź jest znana. I właśnie dlatego czas liznąć temat Big Data w New Space.

W tym artykule chcę podejść do sprawy bardzo ogólnie, fragmentarycznie i technicznie zarazem. Przedstawię kilka miejsc o których wiemy, że wykorzystywane są technologie Big Data oraz jakie dokładnie. Niech będzie to zaledwie zajawką tego olbrzymiego tematu, jakim jest Big Data w kosmosie.  Będziemy go zgłębiać w późniejszych materiałach, ale teraz – po prostu liźnijmy tą fascynującą rzeczywistość;-)

Big Data w JPL (NASA)

Pierwszym kandydatem, którego powinniśmy odwiedzić jest amerykańska NASA, a konkretnie JPL. Jet Propulsion Laboratory to centrum badawcze NASA położone w Kalifornii, które odpowiada za… naprawdę całą masę rzeczy. Niektórzy utożsamiają JPL (JPL – nie JBL, czyli firmy od naprawdę fajnych głośników;-)) z pracą nad łazikami. Słusznie, ale rzeczy które leżą w ich zasięgu jest cała masa.

Według JPL na potrzeby NASA zbierane są setki terabajtów… każdej godziny. Setki Terabajtów! Czy jesteśmy w stanie wyobrazić sobie tak gigantyczne dane? Wystarczy pomyśleć o liczbach, które się generują po miesiącu, dwóch, trzech latach… No jednym słowem: kosmos.

Czego wykorzystują do przetwarzania i analizy takich potwornych ilości danych amerykańscy inżynierowie z NASA? Mamy tu dobrze znane name technologie. Z pewnością jest ich więcej, ja dotarłem do takich jak:

  • Hadoop
  • Spark
  • Elasticsearch + Kibana

Apache Spark w JPL (NASA) – SciSpark

Oczywiście szczególnie mocno, jako freaka na tym punkcie, cieszy mnie użycie Sparka;-). Jeśli chcesz się dowiedzieć na jego temat coś więcej – dobrze będzie jak zaczniesz od mojej serii “Zrozumieć Sparka”. Pytanie – do czego wykorzystywany jest Apache Spark w JPL? Oczywiście do przetwarzania zrównoleglonego danych pochodzących z łazików, satelit i czego tam jeszcze nie mają.

Co ciekawe jednak, inżynierowie  big data w JPL utworzyli osobny program, który nazwali SciSpark. Program jest już niestety zarzucony, ale warto rzucić na niego okiem. Nie znalazłem informacji o przerwaniu prac, jednak wskazują na to przestarzałe treści, oddanie projektu fundacji Apache oraz ostatnie commity z 2018 roku. Na czym jednak polegał SciSpark? Jak wiadomo NASA i generalnie technologie kosmiczne to nie tylko wyprawy na Marsa, Księżyc i badanie czarnych dziur w galaktykach odległych o miliard lat świetlnych. To także, a może przede wszystkim, poznawanie naszego miejsca do życia – Ziemi. I program SciSpark powstał właśnie po to, aby pomagać w przetwarzaniu danych dotyczących naszego środowiska, zmian klimatycznych itd. I tak Big Data pomaga nie tylko w eksploracji “space”, ale także “ze space” pomaga poznawać Ziemię.

SciSpark Technicznie

Wchodząc w temat bardzo technicznie – program został napisany przede wszystkim w Scali. Chociaż twórcy zdają sobie sprawę, z istnienia PySparka oraz tego, że python jest naturalnym językiem Data Sciencystów, uznali że nie będzie odpowiedni ze względów wydajnościowych. Jak mówią sami:

“Ten Spark (w scali – dopisek autora) został  wybrany by uniknąć znanych problemów związanych z opóźnieniami (latency issue) wynikających z narzutu komunikacyjnego spowodowanego kopiowaniem danych z workerów JVM do procesu deamona Pythona w środowisku PySpark. Co więcej – chcemy zmaksymalizować obliczenia w pamięci, a w PySparku driver JVM zapisuje wynik do lokalnego dysku, a następnie wczytuje przez proces Pythona”.

Trzon SciSpark polega na rozszerzeniu sparkowych struktur RDD (Resilient Distributed Dataset) i utworzeniu nowych – sRDD (Scientific Resilient Distributed Dataset). Struktury te mają być dostosowane bardziej do wyzwań naukowców. W jaki sposób dokładnie, z chęcią zgłębię kod SciSparka i napiszę o tym osobny artykuł, dla chętnych geeków;-).

Poza Sparkiem, SciSpark posiada oczywiście całą architekturę systemu – z HDFSem, użytkownikami i interfejsem użytkownika (UI) włącznie. Poniżej ona – dla ciekawskich;-).

Architektura SciSpark – systemu tworzonego przez JPL (NASA).

Co ciekawe, SciSpark został upubliczniony i udostępniony fundacji Apache. Efekt jest oczywisty – teraz także i Ty możesz przeczesać kod, który pierwotnie tworzyli inzynierowie big data z NASA. Publiczne repozytorium znajdziesz tutaj.

Hadoop w NASA

Oczywiście przetwarzanie przetwarzaniem, ale gdzieś trzeba te dane przechowywać. Służy ku temu kolejna świetnie znana nam technologia, czyli Hadoop. Konkretniej być może warto powiedzieć hadoopowym systemie plików, czyli HDFS. To bardzo intuicyjny i dość oczywisty wybór, ponieważ HDFS pozwala rozproszyć pliki na wielu maszynach, co w przypadku tak ogromnych danych jest absolutnie niezbędne.

Prawdopodobnie – tu moja osobista opinia – z biegiem lat będzie trzeba przerzucić się na coś “nowszej generacji” z powodu różnych problemów i ograniczeń HDFSa. Być może dobrym pomysłem byłoby wykorzystanie Apache Ozone. Nie znalazłem informacji czy ktokolwiek w NASA wykorzystuje ten system z przyczyn dość banalnych (pomyśl tylko co wyskoczy gdy wpiszesz “NASA Ozone” w wyszukiwarkę). Wydaje mi się jednak – po pierwszych próbach wykorzystania Ozone, że musi w Wiśle jeszcze trochę wody upłynąć, zanim technologia dojrzeje.

W kontekście storowania plików, warto wspomnieć jakie to dokładnie są pliki. Oczywiście w systemach NASA budowane są liczne ETL’e, a więc i surowe pliki z pewnością są bardzo rozmaitych formatów. Jeśli jednak dane są już przetworzone, to z grubsza zapisywane są w dwóch formatach:

  1. HDF – czyli Hierarchical Data format – to format plików, który został wymyślony już w ubiegłym wieku. Od początku projektowany był tak, żeby mógł przechowywać duże dane. Od początku też – co ważne – wykorzystywany był przez NASA. Nie jest wielką tajemnicą, że tego typu instytucje nie mają zwrotności bolidu F1. Jeśli już do czegoś się przyspawają, pozostanie to z nimi na wieki;-). Więcej na temat HDF można przeczytać w tym dokumencie amerykańskiej agencji.
  2. NetCDF – czyli Network Common Data Form – to z kolei format plików (i związanych z nimi bibliotek), które przeznaczone są do przechowywania danych naukowych. Co ciekawe, pierwotnie NetCDF bazowało na koncepcji Common Data Format opracowanej przez… NASA. Potem jednak NetCDF poszło swoją drogą. To także jest format, który został zapoczątkowany już kilkadziesiąt lat temu.

Elasticsearch w JPL

Zasadniczo problem był następujący: jak w czasie rzeczywistym odtwarzać i przeglądać dane telemetryczne z bardzo, bardzo odległych źródeł. Jednym z najważniejszych był łazik Curiosity. Ten oddalony od nas o 150 milionów mil badał powierzchnie marsa (w rzeczywistości wartość ta dynamicznie się zmienia wraz z krążeniem obu planet wokół słońca). Trzeba było wykorzystać nowoczesne technologie Big Data. Jak może to wyglądać w praktyce? Przykład podaje Tom Soderstrom, Chief Technology and Innovation Officer, and Dan Isla, IT Data Scientist.

“Jeśli udałoby nam się dokładnie przewidzieć parametry termiczne, czas jazdy Curiosity mogłaby wzrosnąć dramatycznie, co mogłoby nas doprowadzić do przełomowych odkryć. I odwrotnie – błąd mógłby poważnie wpłynąć na misję za dwa miliardy dolarów.”

W kibanie można tworzyć wspaniałe środowisko do analizy. I właśnie to skusiło JPL z NASA.

Wcześniej inżynierowie JPL żmudnie zbierali dane i wrzucali je do powerpointa, gdzie potem analitycy mogli je analizować. Trwało to kupę czasu i cóż… z naszej dzisiejszej perspektywy wygląda to wręcz nieprawdopodobnie głupio. Możemy sobie tylko wyobrazić jaką rewolucję wprowadziło zastosowanie technologii Big Data. Konkretnie inżynierowie big data z NASA napisali całą platformę, nazwaną Streams, dzięki której dane mogły przychodzić w czasie “rzeczywistym” (o ile można tak nazywać komunikację z Marsem), a następnie być analizowane i przeszukiwane na bieżąco.

Właśnie w tym przeszukiwaniu i analizowaniu pomógł Elasticsearch wraz ze swoją wierną towarzyszką Kibaną. Dzięki spojrzeniu na problem przeglądania danych telemetrycznych jak na problem wyszukiwania (search problem) można było zaprzegnąć ES i rozwiązać rzeczy do tej pory nierozwiązywalne. Przede wszystkim sprawnie można było ograniczyć zakres poszukiwanych danych i skupić się tylko na tym co trzeba. Można było ładnie wizualizować i przeglądać to co zostało znalezione. Analitycy dostali w swoje ręce narzędzia, o których wcześniej się nie śniło.

JPL to niejedyne miejsce wykorzystujące Big Data w New Space

Zaczynając artykuł byłem przekonany, że zajmie on tylko kawałeczek. Teraz, gdy opowiedziałem o wykorzystaniu Big Data w NASA widzę jak bardzo się pomyliłem. Nie chcę rozwijać materiału jeszcze bardziej, dlatego już teraz zapraszam na drugą część;-). Jeśli chcesz dowiedzieć się jak Big Data wykorzystywana jest w innych obszarach New Space – zapisz się na newsletter lub obserwuj RDF na LinkedIn. Zrób to koniecznie i razem twórzmy polską społeczność Big Data!

 

Loading

Zrozumieć sparka: uruchamianie joba na serwerze (spark submit)

Zrozumieć sparka: uruchamianie joba na serwerze (spark submit)

Przygotowywałem się ostatnio do kolejnego szkolenia ze Sparka. Chciałem dla swoich kursantów przygotować solidną rozpiskę tego co i jak robić w spark submit. Niemałe było moje zdziwienie, gdy dokumentacja ani stackoverflow nie odpowiedziały poważnie na moje prośby i błagania. Cóż… czas zatem samodzielnie połatać i przedstawić od podstaw kwestię spark submit – czyli tego jak uruchamiać nasze aplikacje (joby) sparkowe na serwerze. Zapraszam!

Spark Submit – podstawy

Zacznijmy od podstaw. Kiedy piszemy naszą aplikację Sparkową, robimy to w swoim IDE. Przychodzi jednak moment, w którym musimy wysłać ją na serwer. Pytanie – jak uruchamiać poprawnie joba sparkowego na klastrze, który zawiera wiele nodów?

Służy temu polecenie spark submit. Uruchamiając je musimy mieć jedynie dostęp do naszego pliku jar lub pliku pythonowego. Wszystko poniżej będę robił na plikach jar. Musimy to polecenie wykonywać oczywiście na naszym klastrze, gdzie zainstalowane są biblioteki sparkowe. Dobrze, żebyśmy mieli także dostęp do jakiegoś cluster managera – np. YARNa.

W ramach spark submita musimy określić kilka podstawowych rzeczy:

  1. master – czyli gdzie znajduje się master. W rzeczywistości wybieramy tutaj także nasz cluster manager (np. YARN).
  2. deploy-mode – client lub cluster. W jakim trybie chcemy uruchomić nasz job.
  3. class – od jakiej klasy chcemy wystartować
  4. plik wykonywalny – czyli np. jar.

To są podstawy które muszą być. To jednak dopiero początek prawdziwej zabawy z ustawieniami joba w spark submit. W samym środku możemy wskazać znacznie więcej ustawien, dzięki którym nadamy odpowiedni kształt naszej aplikacji (jak choćby kwestia pamięcie w driverze czy w executorach). Mały hint: chociaż można za każdym razem wpisywać wszystko od początku, polecam zapisać spark submit w jednym pliku, (np. run.sh) a następnie uruchamiać ten konkretny plik. To pozwoli nie tylko archiwizować, ale przede wszystkim wygodnie korzystać z bardzo długiego polecenia.

Poniżej pokazuję przykład spark submit do zastosowania.

spark-submit \
--class App \
--master yarn \
--deploy-mode "client" \
--driver-memory 5G \
--executor-memory 15G \
rdfsuperapp.jar

Jak dokładnie rozpisać ustawienia joba?

Tak jak napisałem, opcji w spark submit jest naprawdę, naprawdę wiele. Każda z nich znaczy co innego (choć występuje dużo podobieństw) i poniżej zbieram je wszystkie (lub większość) “do kupy”. Robię to m.in. dlatego, że chcąc znaleźć znaczenie wielu z nich, musiałem nurkować w różnych źródłach. Zatem nie dziękuj – korzystaj;-).

Poniżej najczęstsze ustawinia dla spark submit.

–executor-cores Liczba corów przydzielonych do każdego executora. UWAGA! Opcja dostępna tylko dla Spark Standalone oraz YARN
–driver-cores Liczba corów dla drivera. Dostępne tylko w trybie cluster i tylko dla Spark Standalone oraz YARN
–files Pliki które dołączamy do sparka z lokalnego systemu. Potem pliki te są przesłane każdemu worker nodowi, dzieki czemu możemy je wykorzystać w kodzie.
–verbose Pokazuje ustawienia przy starcie joba.
–conf ustawienia konfiguracyjne (więcej o tym w rozdziale poniżej)
–supervise Jeśli podane, restartuje drivera w przypadku awarii. UWAGA! Dostępne tylko dla Spark Standalone oraz Mesos.
–queue Określenie YARNowej kolejki na której ma być startowany. UWAGA! Dostępne tylko dla YARN.
–packages Lista oddzielonych przecinkami adresów mavena dla plików jar, które chcemy przyłączyć do drivera lub executorów. Najpierw będzie przeszukiwał lokalne repo mavena, potem centralne oraz zdalne repozytoria podane w –repositories.Format: groupId:artifactId:version.
–exclude-pachages Lista dependencji, które mają być excludowane z listy –packages. Kolejne elementy oddzielone są przecinkiem i mają nastepujący format: groupId:artifactId.
–repositories Lista oddzielonych przecinkami repozytoriów do których uderzamy przy okazji ustawienia –packages.

Poza ustawieniami służącymi do odpowiedniego zasubmitowania joba, można przekazać także kilka kwestii konfiguracyjnych. Dodajemy je za pomocą ustawienia –conf w spark-submit. Następnie podajemy klucz-wartość. Może być podana bezpośrednio lub przy użyciu cudzysłowów, np:

--conf spark.eventLog.enabled=false
--conf "spark.executor.extraJavaOptions=-XX:+PrintGCDetails -XX:+PrintGCTimeStamps"

Jak uruchamiać Sparka za pomocą YARNa i bez?

Spark Submit służy nam do tego, żeby uruchamiać naszego joba na klastrze. Możemy do tego wykorzystać jakiś zewnętrzny cluster manager, lub zrobić to przy pomocy Spark Standalone. Spark Standalone to taki tryb Sparka, w którym jest on postawiony na klastrze jako realnie działający serwis (z procesami itd). Cluster managerem może być YARN ale może być to także Mesos, Kubernetes i inne.

Za każdym razem gdy chcemy użyć sparka, musimy wskazać mastera (poprzez wlaśnie ustrawienie master w spark submit). Jeśli korzystamy z YARNa, sprawa jest banalnie prosta – do ustawienia master przekazujemy po prostu yarn. Reszta jest zrobiona.

Rzecz wygląda ciut trudniej, gdy chcemy wskazać za pomocą innych cluster managerów.

  1. Spark Standalone: spark://<ip>:<port> np –master spark//207.184.161.138:7077
  2. Kubernetes: –master k8s://https://<k8s-apiserver-host>:<port>
  3. Mesos: –master mesos://<host>:<port> (port: 5050)
  4. YARN: –master yarn

W niektórych dystrybucjach należy dodać także parę rzeczy, jednak są to już dodatkowe operacje. Jeśli będzie chęć – z przyjemnościa pochylę się nad bardziej szczeółowym porównaniem wyżej wymienionych cluster managerów;-).

Głodny Sparka?

Mam nadzieję, że rozwiązałem Twój problem ze zrozumieniem Spark Submit oraz sensu poszczególnych właściwości. Jeśli jesteś wyjątkowo głodny/a Sparka – daj znać szefowi. Przekonaj go, żeby zapisał Ciebie i Twoich kolegów/koleżanki na szkolenie ze Sparka. Omawiamy całą budowę od podstaw, pracujemy dużo i intensywnie na ciekawych danych, a wszystko robimy w miłej, sympatycznej atmosferze;-) – Zajrzyj tutaj!

A jeśli chcesz pozostać z nami w kontakcie – zapisz się na newsletter lub obserwuj RDF na LinkedIn. Koniecznie, zrób to i razem twórzmy polską społeczność Big Data!

 

Loading
Zrozumieć Sparka: cache vs persist

Zrozumieć Sparka: cache vs persist

W naszej podróży po Sparku musiała przyjść pora na ten przystanek. “Cache vs persist” to absolutnie podstawowe pytanie na każdej rozmowie rekrutacyjnej ze Sparka. Co ważniejsze – to naprawdę przydatne narzędzie dla każdego inżyniera big data, który posługuje się tym frameworkiem! Dzięki niemu przyspieszysz swojego joba, zmniejszysz ryzyko nieoczekiwanego fuckupu. No i najważniejsze – podczas code review pokażesz, że znasz się na rzeczy i nie jesteś tu z przypadku. Żartuję oczywiście.

Nie traćmy więc czasu i sprawdźmy co to za dobrodziejstwo, owe osławione cache i persist. Całą serię “zrozumieć sparka” poznasz tutaj.

Krótka powtórka – akcje i transformacje

Zanim przejdziemy do sedna sprawy – najpierw krótkie przypomnienie jak działa Spark – w pigułce. Aby to zrozumieć, fundamentalne są trzy rzeczy:

  1. Spark działa na swoich własnych strukturach, “kolekcjach”. To przede wszystkim RDD, które przypominają “listy” lub “tablice” (nie w konkrecie, natomiast grunt że nie jest to kolekcja typu mapa), ale także Datasety i Dataframy (które są aliasem dla Dataset[Row]). Napisałem, że “przede wszystkim”, ponieważ pod spodem DSów i DFów leżą właśnie RDD. To na tych strukturach musimy pracować, żeby całość ładnie działała.
  2. Na tych strukturach możemy wykonywać dwa typy operacji:
    • transformacje – czyli operacje, które wykonują pewne obliczenia, natomiast nie są wykonywane w momencie ich wywołania w kodzie.
    • akcje – operacje, które zwykle są jakimś rodzajem operacji wyjścia”. I co istotne – w tym momencie wykonywane są wszystkie transformacje po kolei.
  3. Takie podejście lazy evaluation zastosowane jest, aby Spark mógł odpowiednio zoptymalizować te operacje. Spisuje je w drzewku DAG (Directed Acyclic Graph – graf skierowany, acykliczny).

Tutaj pojawia się właśnie podstawowy problem – może się okazać, że akcji będziemy mieli kilka. I za każdym razem Spark będzie musiał wykonywać te same transformacje. To oczywiście powoduje, że job staje się niewydajny, a moc obliczeniowa jest niepotrzebnie przepalana.

Cache i persist w Sparku

I właśnie tutaj wkracza cachowanie. Możemy na danym zbiorze danych wywołać .cache() lub .persist() aby zachować go w pamięci. Gdyby potem pojawiło się kilka akcji, Spark będzie pobierał dane z konkretnego punktu, a nie procesował je wszystkimi transformacjami od początku.

cache vs persist

Aby to zrobić, mamy do dyspozycji dwie funkcje: cache() oraz persist(). Cóż – prawdę mówiąc jest jeszcze trzecia (.checkpoint()) ale na ten moment ją zostawimy;-). Tak więc powtórzmy – cache i persist pozwalają “zatrzymać” dataset w kodzie. Od momentu wywołania na naszych danych (RDD, Dataset lub Dataframe) cache lub persist – wywołanie akcji nie będzie skutkowało ponownym przeliczaniem transformacji, które były napisane przed tym punktem.

Pora jednak wyjaśnić sobie różnicę między cache a persist. Otóż – mówiąc prosto – persist pozwala nam ustawić poziom na jakim chcemy przechowywać dane (storage level), natomiast cache ma domyślne wywołanie. Tak naprawdę pod spodem cache wywołuje… właśnie persist!

Jaki storage level ma cache? To zależy od tego czy wywoujemy go na RDD czy Dataset/Dataframe. Jeśli RDD – domyślnie mamy MEMORY_ONLY, natomiast w przypadku Datasetów – MEMORY_AND_DISK.

Kod cache i persist dla Dataset i Dataframe

Jak wygląda wywołanie w kodzie? Przede wszystkim, jeśli wywołujemy cache() na Dataframe, to jest to najzwyklejszy w świecie alias dla wywołania funkcji persist(). I nie zmieniło się to od dawna. Poniżej kod z “bebechów” Sparka 2.4.3.

/**
   * Persist this Dataset with the default storage level (`MEMORY_AND_DISK`).
   *
   * @group basic
   * @since 1.6.0
   */
  def cache(): this.type = persist()

Jak wygląda natomiast persist()? W tym przypadku mamy do czynienia z dwoma funkcjami: pierwsza to funkcja, która przyjmuje argument typu StorageLevel, natomiast druga jest funkcją bez argumentów. Wywołuje ona funkcję cacheQuery() znajdującą się w CacheManager, która dokonuje całej “magii”. Jeśli nie podamy jej Storage Level, zostanie przypisany standardowy – czyli “MEMORY_AND_DISK”.

Poniżej implementacje obu funkcji persist().

def persist(): this.type = {
    sparkSession.sharedState.cacheManager.cacheQuery(this)
    this
  }

def persist(newLevel: StorageLevel): this.type = {
    sparkSession.sharedState.cacheManager.cacheQuery(this, None, newLevel)
    this
  }

Kod cache i presist dla RDD

W kontekście RDD cache() także jest aliasem dla persist(). Znów mamy także do czynienia z dwiema funkcjami persist() – jedną bezargumentową, drugą przyjmującą argument typu StorageLevel. Tym razem jednak różnica jest taka, że standardowym, domyślnym poziomem jest “MEMORY_ONLY” (nie jak w przypadku datasetów “MEMORY_AND_DISK”).

Można nawet dostrzec “niezwykle elegancki” komentarz, który pokusił się zostawić w samym środku funkcji jeden z inżynierów tworzących Sparka;-).

def persist(newLevel: StorageLevel): this.type = {
    if (isLocallyCheckpointed) {
      // This means the user previously called localCheckpoint(), which should have already
      // marked this RDD for persisting. Here we should override the old storage level with
      // one that is explicitly requested by the user (after adapting it to use disk).
      persist(LocalRDDCheckpointData.transformStorageLevel(newLevel), allowOverride = true)
    } else {
      persist(newLevel, allowOverride = false)
    }
  }

Storage levels

Z implementacji funkcji persist() wiemy, że kluczową sprawą jest Storage Level – czyli poziom na jakim chcemy przechowywać nasze dane. Oto jakie poziomy możemy wybrać:

  1. MEMORY_ONLY – przechowuje RDD lub Datasety jako deserializowane obiekty w pamięci JVM. Jeśli nie ma wystarczająco dużo dostępnej pamięci, niektóre partycje nie zostaną zapisane i zostaną potem ponownie obliczone w razie potrzeby.
  2. MEMORY_ONLY_2 – To samo co MEMORY_ONLY, natomiast każda partycja jest replikowana na dwa nody.
  3. MEMORY_ONLY_SER – to samo co MEMORY_ONLY, natomiast przechowuje dane jako serializowane obiekty w pamięci JVM. Potrzebuje mniej pamięci niż [1].
  4. MEMORY_ONLY_SER_2 – analogicznie.
  5. MEMORY_AND_DISK – w tym przypadku dane są przechowywane jako deserializowane obiekty w pamięci JVM. W tym przypadku jednak, w przeciwieństwie do [1], jeśli nie wystarczy miejsca, nadmiarowe partycje są przechowywane na dysku. To prowadzi do mniejszej wydajności (wolniejszy proces), ponieważ angażujemy tu operacje I/O.
  6. Analogicznie do 1-4 mamy także MEMORY_AND_DISK_SER, MEMORY_AND_DISK_2, MEMORY_AND_DISK_SER_2.
  7. DISK_ONLY – w tym przypadku dane są przechowywane jedynie na dysku. Oczywiście najwolniejszy wariant.
  8. DISK_ONLY_2 – analogicznie, tworzone są replikacje.

Zostań prawdziwym sparkowym wojownikiem

Jeśli dotarłeś do tego punktu, to prawdopodobnie jesteś naprawdę zainteresowany rozwojem w Sparku. Trzeba uczciwie powiedzieć, że to złożona, duża technologia. Jeśli chcesz zgłębić ją bardziej – przekonaj swojego szefa, żeby zapisał Ciebie i Twoich kolegów/koleżanki na szkolenie ze Sparka. Omawiamy całą budowę od podstaw, pracujemy dużo i intensywnie na ciekawych danych, a wszystko robimy w miłej, sympatycznej atmosferze;-) – Zajrzyj tutaj!

A jeśli chcesz pozostać z nami w kontakcie – zapisz się na newsletter lub obserwuj RDF na LinkedIn. Koniecznie, zrób to i razem twórzmy polską społeczność Big Data!

 

Loading
3 kroki do przodu: jak Big Data może pomóc Polsce w opanowaniu inflacji?

3 kroki do przodu: jak Big Data może pomóc Polsce w opanowaniu inflacji?

Inflacja po raz pierwszy (od dawna) weszła “pod strzechy” – nie jest już jedynie tematem dyskusji eksperckich. Wręcz przeciwnie – od kilku miesięcy jest bohaterką pierwszych stron gazet w całym kraju – z brukowcami włącznie. Powodem jest znaczne przyspieszenie utraty wartości waluty, co w przypadku naszej historii wzbudza szczególnie nieprzyjemne skojarzenia. Dodatkowo niektórzy zarzucają stronie rządowej, że oficjalna inflacja jest zaburzona.

Chciałbym dzisiaj zaproponować pewne rozwiązanie, które pomogłoby nam w analizie inflacji, a co za tym idzie – w odpowiedniej kontroli nad nią. Wszystko co poniżej to pewna ogólna wizja, która może posłużyć jako inspiracja. Jeśli jest chęć i zapotrzebowanie, bardzo chętnie się w tą wizję zagłębię architektonicznie i inżyniersko. Zachęcam także do kontaktu, jeśli TY jesteś osobą zainteresowaną tematem;-).

Krótka lekcja: jak liczona jest inflacja?

Czym jest inflacja?

Zanim przejdziemy do rozwiązania, zacznijmy od problemu. Czym jest inflacja i jak liczy ją GUS? Przede wszystkim najważniejsze to zrozumieć, że inflacja to spadek wartości pieniądza w czasie. Dzieje się tak w sposób praktyczny poprzez wzrost cen. Za ten sam chleb, wodę – musimy zapłacić więcej. I teraz to najważniejsze: inflacja jest inna dla każdego z nas. Każdy z nas ma bowiem inny portfel.

Jeśli zestawimy samotnego programistę oraz rodzinę wielodzietną, gdzie Tata zarabia jako architekt a Mama jako tłumaczka, ich budżety bedą zupełnie inne. Nawet jeśli zarabiają podobne kwoty, w rodzinie większy udział prawdopodobnie będzie na pieluchy, przedmioty szkolne i parę innych rzeczy. W przypadku młodego singla z solidną pensją, do tego z rozrywkowym podejściem do życia, znacznie większy procent budżetu zajmie alkohol, hotele, imprezy itd. Jeśli ceny alkoholi pójdą w górę o 40%, dla niektórych inflacja będzie nie do zniesienia, dla innych z kolei może nie zostać nawet zauważona.

Jak liczone jest CPI (inflacja konsumencka)?

Aby zaradzić tego typu problemom, GUS wylicza coś takiego jak CPI (Consumer Price Index) – indeks zmiany cen towarów i usług konsumpcyjnych. W skrócie mówimy inflacja CPI, czyli inflacja konsumencka. Warto zaznaczyć tutaj jeszcze, że zupełnie inna może być inflacja odczuwana na poziomie budżetów firm (a właściwie dla różnych firm jest także oczywiście różna).

Nie będziemy się tutaj wgryzać zbyt mocno w to jak dokładnie wylicza się inflację CPI. Skupimy się jedynie na paru najważniejszych rzeczach, które przydadzą nam się do późniejszej odpowiedzi na nasz inflacyjny problem;-). Dla zainteresowanych polecam solidniejsze omówienia:

  1. Najpierw u źródła – “Co warto wiedzieć o inflacji?” przez GUS.
  2. “Ile naprawdę wynosi inflacja?” Marcin Iwuć
  3. “GUS zaniża inflację? Ujawniamy!” – mBank
Koszyk inflacyjny 2021. Autor: Pawelmhm

Na nasze potrzeby powiedzmy sobie bardzo prosto, w jaki sposób GUS liczy CPI. Potrzeba do tego podstawowej rzeczy, czyli koszyka inflacyjnego. Taki koszyk to grupy towarów, które podlegają badaniu. GUS oblicza to na podstawie ankiet wysyłanych przez 30 000 osób. Już tutaj powstaje pewien problem – ankiety te mogą być wypełniane nierzetelnie.

Następnie, jeśli mamy już grupy produktów, musimy wiedzieć jak ich ceny zmieniają się w skali całego kraju. Aby to zrobić, wyposażeni w tablety ankieterzy, od 5 do 22 dnia każdego miesiąca, ruszają do akcji – a konkretnie do wytypowanych wcześniej punktów (np. sklepów spożywczych) w konkretnych rejonach. W 2019 roku badanie prowadzono w 207 rejonach w całej Polsce.

Big Data w służbie jej królew… to znaczy w służbie Rządu RP

Taka metodologia prowadzi do bardzo wielu wątpliwości. Badanie GUSu zakrojone jest na bardzo szeroką skalę. Mimo to jednak wciąż są to jedynie wybrane gospodarstwa oraz wybrane punkty sprzedaży. Chcę tutaj podkreślić, że nie podejrzewam naszych statystyków o manipulacje. Może jednak dałoby się zrobić tą samą pracę lepiej, bardziej precyzyjnie i znacznie mniejszym kosztem?

Cyfryzacja paragonowa – czyli jak Rząd przenosi nasze zakupy do baz danych?

Zanim przejdziemy dalej, powiedzmy najpierw coś, z czego być może większość z nas sobie nie zdaje sprawy. Ostatnie lata to stopniowe przechodzenie z tradycyjnych kas fiskalnych na kasy wirtualne oraz online. Na potrzeby artykułu nie będę wyjaśniał różnic. To co nas interesuje to fakt, że oba typy kas, zakupy raportują bezpośrednio do Rządu. Na ten moment objęta jest tym stanem rzeczy gastronomia, ale docelowo ma to objąć (wedle mojej wiedzy) także inne sektory posługujące się paragonami.

Jak zbudować system liczący inflację?

Przenieśmy się mentalnie do momentu, w którym każda, lub niemal każda sprzedaż jest odnotowywana przez państwo i zapisywana w tamtejszej bazie danych (najprawdopodobniej nierelacyjnej). Można to wykorzystać, aby zbudować system, który pozwoli nam liczyć inflację pozbawioną potrzeby wysyłania armii ankieterów. Co więcej – pozwoli to zrobić dokładniej oraz da nam potężne narzędzie analityczne!

Bazy danych / Storage

Przede wszystkim – zakładam, że wszystkie dane trzymane są w jakiejś nierelacyjnej bazie danych – na przykład w Apache HBase. Może to być jednak także rozproszony system plików, jsk HDFS. W takiej bazie powinny być trzymane wszystkie dane dotyczące transakcji – paragony, faktury, JPK itd. Osobną sprawą pozostają informacje dotyczące firm i inne dane, które są bardziej “ogólne” – dotykają mniejszej liczby podmiotów i nie są tak detaliczne.

W nowoczesnym systemie do liczenia inflacji Spark odegrałby kluczową rolę

Te dane, ze względu na niewielką liczbę i bardzo klarowną strukturę, można trzymać w bazie relacyjnej (np. PostgreSQL). Można jednak także jako osobną tabelę HBase, choć z przyczyn analitycznych (o których potem) znacznie lepiej będzie zrobić to w bazie relacyjnej. Można także zastosować rozwiązanie hybrydowe – wszystkie dane dotyczące firm trzymać w bazie nierelacyjnej, jako swoistym “magazynie”, natomiast pewną wyspecyfikowaną, odchudzoną esencję – w bazie relacyjnej.

Dodatkowo zakładam, że koszyk inflacyjny jest już wcześniej przygotowany. Da się ten proces uprościć poprzez informatyzację ankiet – jest to już zresztą robione (według wiedzy jaką mam). Taki koszyk można trzymać w bazie relacyjnej, ze względu na relatywnie niewielką liczbę danych (W 2021 roku zawierał on 12 grup produktów. Nawet jeśli w każdej z nich byłoby parę tysięcy produktów, liczby będą  sięgać maksymalnie kilkudziesięciu tysięcy, niezbyt rozbudowanych rekordów).

W dalszej części dodam jeszcze możliwość tworzenia kolejnych koszyków i w takiej sytuacji prawdopodobnie należałoby już je wydzielić do osobnej bazy nierelacyjnej. W dalszym ciągu jednak ogólne adnotacje mogłyby pozostać w bazie relacyjnej (tak, żeby można było np. sprawnie wyciągnąć dane z HBase po rowkey, czyli id).

Jeśli zdecydujemy się na zastosowanie bazy row-key, jak HBase, uważam że i tak zaistnieje potrzeba wykorzystania HDFS (może być tak, że w HBase będzie wygodniej pierwotnie umieszczać pliki paragonowe). Będziemy tu umieszczać kolejne etapy przetworzonych danych z konkretnych okresów.

Jeszcze inną opcją jest zastosowanie Apache Kudu, który mógłby nieco zrównoważyć problemy HBase i HDFS i zastąpić oba byty w naszym systemie. Jak widać, opcji jest dużo;-)

Przygotowanie danych

Kiedy mamy już dane zebrane w przynajmniej dwóch miejscach, powinniśmy je przygotować. Same z siebie stanowią jedynie zbiór danych, głównie tekstowych. W drugim  etapie należy te dane przetworzyć, oczyścić i doprowadzić do postaci, w jakiej ponownie będziemy mogli dokonać już finalnej analizy inflacji.

 

Finałem tego etapu powinny być dane, które będą pogrupowane tak, żebyśmy mogli je później wykorzystać. Wstępna, proponowana struktura wyglądać może następująco:

  1. Okres badania
    1. Grupa produktów
      1. Punkt sprzedaży
        1. towar

Musimy więc wyciągnąć surowe dane (z HBase), przetworzyć je, a następnie zapisać jako osobny zestaw – proponuję tu HDFS. Jak to uczynić? Możemy do tego celu wykorzystać Apache Spark oraz connector HBase Spark przygotowany przez Clouderę. Następnie dane muszą być poddane serii transformacji, dzięki którym dane:

  • Zostaną wydzielone jako osobne paragony
  • Zostaną podzielone na produkty
  • Poddane będą oczyszczeniu z wszelkich “śmieci” uniemożliwiających dalszą analizę
  • Wykryta zostanie grupa produktów dla każdego z nich
  • Pogrupowane zostaną po grupie produktów oraz okresie

Na końcu dane zapisujemy do HDFS. Wstępna struktura katalogów:

  1. Dane przygotowane
    1. Dane całościowe
      1. okres
        1. Tutaj umieszczamy plik *.parquet lub *.orc

Liczenie inflacji

Skoro mamy już przygotowane dane, czas policzyć inflację. Do tego celu także wykorzystamy Apache Spark, dzięki któremu możemy w zrównoleglony sposób przetwarzać dane. W najbardziej ogólnym kształcie sprawa wygląda dość prosto:

  1. Łączymy się z bazą danych (relacyjną), w której trzymamy konkretny koszyk
  2. Wybieramy okres za jaki chcemy policzyć inflację
  3. Pobieramy dane z HDFS/Kudu, które okresem odpowiadają [2].
  4. Wybieramy grupy produktów zgodne z koszykiem [1]
  5. Przeliczamy inflację za pomocą danych, które są już solidnie przygotowane.

I teraz ważne: efekt zapisujemy do relacyjnej bazy danych.

Analiza

Czemu akurat do relacyjnej bazy danych? Odpowiedź wydaje się oczywista:

  1. Dane będą niewielkie – choć od raz umożna powiedzieć, że z naszego procesu można wycisnąć więcej niż tylko wynik 6.8% 😉 – jest też sporo rzeczy przy okazji, takie jak jakie produkty wzrosły najmocniej, w jakich regionach, co ma największą zmiennośći itd.
  2. Dane będą solidnie ustrukturyzowane
  3. Dane umieszczone w relacyjnych bazach pozwalają na znacznie lepszą i prostszą analizę.

I właśnie ten trzeci punkt powinien nas zainteresować najmocniej. Można bowiem na klastrze zainstalować jakieś narzędzie BI, spiąć z bazą i… udostępnić analitykom. Takim narzędziem może być (znów open sourcowy) Apache Superset. Przynajmniej na dobry początek. W drugim rzucie należałoby się pokusić o zbudowanie dedykowanej aplikacji analitycznej. To jednak można zostawić na  później. Na etap, w którym analitycy będą już zaznajomieni z systemem i będą mogli włączyć się w czynny proces budowy nowego narzędzia.

Rozwój

Wyżej opisałem podstawowy kształt systemu do badania inflacji. Warto jednak nie zatrzymywać się na tym i pomyśleć jak można tą analizę wynieść na wyższy poziom. Podstawową prawdą na temat inflacji jest to, że każdy ma swoją, więc nie da się dokładnie obliczać jak pieniądz traci na wartości. Cóż… dlaczego nie możnaby tego zmienić? Wszak mając do dyspozycji WSZYSTKIE (oczywiście upraszczając) dane, można zrobić “nieskończenie wiele” koszyków inflacyjnych.

Czym mogłoby być te koszyki inflacyjne? Kilka propozycji.

  1. Dlaczego nie wykorzystać rozwoju technologii Big Datowych do podnoszenia świadomości finansowej oraz obywatelskiej Polaków? Niech każdy będzie mógł na specjalnym portalu wyklikać swój własny koszyk i regularnie otrzymywać powiadomienia dotyczące swojej własnej inflacji. Takie podejście byłoby bardzo nowatorskie i z pewnością wybilibyśmy się na tle innych państw.
  2. Koszyki mogą powstawać dla różnych grup społecznych. Dzięki temu można będzie dokładniej badać przyczyny rozwarstwienia społecznego, niż jedynie osławiony współczynnik Giniego, czy także inne, powiedzmy sobie szczerze – skromne narzędzia, na podstawie których wyciągane są bardzo mocne dane.
  3. Koszyki dla firm – oczywistym jest, że firmy mają znacząco różny koszyk od ludzi. Jest oczywiście inflacja przemysłowa (PPI), natomiast dotyczy ona produkcji przemysłowej (i to w dośćwąskim zakresie). Dzięki wyborowi produktów w “naszym systemie” można będzie obliczyć także jak bardzo wartość pieniądza spada dla różnych rodzajów firm.

Potencjalne korzyści

Powyżej opisałem przykładowy system, który pozwoliłby nam wynieść analizę inflacji na zupełnie inny poziom. Poniżej chciałbym zebrać w jedno miejsce kilka najważniejszych korzyści, jakie niosłyby takie zmiany:

  1. Mniejsze koszty – cykliczne uruchamianie jobów mających na celu sprawdzenie inflacji to koszt znacznie mniejszy, niż utrzymywanie armii ankieterów.
  2. Dokładniejsza inflacja – precyzja liczenia inflacji weszłaby na zupełnie inny poziom. Oczywiście na początku należałoby przez kilka lat liczyć w obu systemach, aby sprawdzić jak bardzo duże są różnice.
  3. Różne modele inflacji – a więc koszyki o których pisałem powyżej, które spowodują, że przestanie być prawdziwa teza o tym, że “liczenie prawdziwej inflacji nie jest możliwe”.
  4. Regionalizacja inflacji – Inflacja inflacji nie jest równa. Zupełnie inaczej ceny mogą się kształtować w różnych województwach. Również i to mógłby liczyć “nasz system”.
  5. Większe możliwości analityczne – stopy nie są jedynym narzędziem, który można użyć w walce z inflacją. Ekonomiści wskazują, że poza wysokością stóp procentowych także inne czynniki wpływają na inflację. Są to m.in. skala dodruku pieniądza, rozwój świadczeń socjalnych, regulacje gospodarcze czy wysokość podatków pośrednich. Dzięki Big Datowemu systemowi, Rząd zyskałby znacznie większe możliwości analityczne do śledzenia wpływu swoich zmian na gospodarkę.
  6. Wyższe morale i poczucie, że państwo gra “w pierwszej lidze” – unowocześnia swoje działanie na poziom niespotykany w innych krajach.

Potencjalne zagrożenia

Rozwój zaawansowanych systemów to oczywiście także zagrożenia. O tych najważniejszych poniżej:

  1. Możliwości analityczne muszą wiązać się z większą inwigilacją – i jest to chyba największy problem. Im większe możliwości analizy chcemy sobie zafundować, tym głębiej trzeba zinfiltrować życie obywateli. Oczywiście przed infiltracją współcześnie uciec się nie da, ale należy zawsze mieć na uwadze mądre wyważanie.
  2. Koszt utrzymania systemu – To system zbierania bardzo dużych danych i analizy ich. Wymagać będzie zaawansowanych klastrów obliczeniowych oraz odpowiedniego zespołu administracyjnego. Na pensjach – dodajmy – zdecydowanie rynkowych, nie urzędniczych.
  3. Kwestia bezpieczeństwa – czasami zapominamy, że informatyzacja państwa to problem w bezpieczeństwie danych. Jeśli jesnak dane dotyczące zakupów i tak miałyby być zbierane – czemu ich nie wykorzystać?

Big Data to nasza wielka szansa – także w sektorze rządowym

Jesteśmy Państwem, które w wielu miejscach wiecznie próbuje nadgonić resztę (choć w wielu także tą resztę przegoniło). Big Data może pozwolić na działać lepiej, szybciej, precyzyjniej i… taniej. Wykorzystajmy tą szansę. Za nami poglądowy pomysł na to jak zbudować jeden z takich systemów, które mogłyby pomóc nam budować nowoczesne państwo. Jeśli chcesz dowiadywać się o innych pomysłach, które ukazują Big Data w odpowiednim kontekście, zapraszam na nasz profil LinkedIn oraz do zapisania się na newsletter RDF. Do zobaczenia na szlaku BD!;-)

 

Loading
Zrozumieć Sparka: czym różnią się podobne mechanizmy? Distinct vs dropDuplicates i inne.

Zrozumieć Sparka: czym różnią się podobne mechanizmy? Distinct vs dropDuplicates i inne.

Pracując ze sparkiem bardzo często spotykamy mechanizmy, które budzą naszą konsternację z powodu bardzo dużego podobieństwa. Czy każde z tych mechanizmów ma inne działanie? Jeśli tak, to jakie są praktyczne różnice? Postanowiłem zanurkować nieco w kod i sprawdzić najbardziej popularne funkcje. Zapraszam!

Aha… to oczywiście artykuł w serii “Zrozumieć Sparka”. Poprzednio omawiałem joiny – zainteresowanych zapraszam tutaj.

Porządkujemy – czyli orderBy vs sort

Pierwszy mechanizm, który bierzemy na tapet to sortowanie. Spark daje nam przynajmniej dwie funkcje, które spełniają to zadanie: orderBy() oraz sort(). Czym się różnią?

Tutaj sprawa jest banalnie prosta – orderBy() to po prostu alias dla sort(). Mówi nam o tym poniższy fragment kodu:

/**
   * Returns a new Dataset sorted by the given expressions.
   * This is an alias of the `sort` function.
   *
   * @group typedrel
   * @since 2.0.0
   */
  @scala.annotation.varargs
  def orderBy(sortExprs: Column*): Dataset[T] = sort(sortExprs : _*)

Mamy dwie funkcje orderBy() – obie wywołują sort().

Zmieniamy nazwy kolumn: withColumnRenamed vs alias

Załóżmy, że mamy dataframe i nie pasuje nam nazwa jednej z kolumn. W takiej sytuacji mamy do dyspozycji dwie – cztery metody: withColumnRenamed(), alias(), as() oraz name(). Zajmijmy się najpierw różnicami między withColumnRenamed oraz alias.

  1. Zwracany typ:
    • alias() to funkcja zwracająca typ Column,
    • withColumnRenamed() zwraca Dataset[Row] (czyli Dataframe).
  2. Co dzieje się w środku:
    • alias() jest jedynie… aliasem dla funkcji name(). Nie robi nic innego. O tym co robi name() rozdział niżej.
    • withColumnRenamed() ma logikę, która zmienia schemat. Warto dodać, że sama zmiana nazwy kolumny dzieje się poprzez wykorzystanie funkcji as(). To – zdradzę przedwcześnie – jest alias dla aliasu. A alias jest aliasem dla name(). WOW! No nic… poniżej podrzucam jak zbudowana jest funkcja withColumnRenamed():

      def withColumnRenamed(existingName: String, newName: String): DataFrame = {
          val resolver = sparkSession.sessionState.analyzer.resolver
          val output = queryExecution.analyzed.output
          val shouldRename = output.exists(f => resolver(f.name, existingName))
          if (shouldRename) {
            val columns = output.map { col =>
              if (resolver(col.name, existingName)) {
                Column(col).as(newName)
              } else {
                Column(col)
              }
            }
            select(columns : _*)
          } else {
            toDF()
          }
        }
  3. Kiedy stosować:
    • alias() stosujemy wtedy, gdy możemy odwołać się do konkretnej kolumny – klasycznym przykładem jest zastosowanie funkcji select(). Tak jak poniżej:
      peopleDF.select(col("name").alias("firstName"))
    • withColumnRenamed() wywołujemy na Dataframe. W efekcie dostajemy nowy Dataframe ze zmienioną nazwą kolumny. Wygląda to tak jak poniżej:
      peopleDF.withColumnRenamed("name", "firstName")

 

Zmienianie nazw ciąg dalszy – as vs alias vs name

Skoro już znamy różnice między withColumnRenamed oraz alias, warto przejść do kolejnych meandrów sparkowego labiryntu. Tym razem poszukajmy czym różnią się od siebie trzy funkcje: as(), alias() oraz name().

Otóż, tym razem sprawa jest prostsza. Albo i bardziej zaskakująca, sam(/a) już musisz to rozstrzygnąć. Otóż – nie ma różnic. A konkretniej as() jest aliasem dla name(), tak samo jak alias() jest aliasem dla name().

Funkcja name() natomiast jest dość prostym mechanizmem. Jeśli obecna kolumna ma jakieś metadane powiązane z nią, zostaną propagowane na nową kolumnę. W  implementacji użyty jest Alias – warto jednak pamiętać, że nie chodzi tu o funkcję alias(), a o case classę Alias. Implementacja funkcji name() poniżej:

def name(alias: String): Column = withExpr {
    expr match {
      case ne: NamedExpression => Alias(expr, alias)(explicitMetadata = Some(ne.metadata))
      case other => Alias(other, alias)()
    }
  }

Usuwamy duplikaty w Sparku – distinct vs dropDuplicates

Prędzej czy później przychodzi taki moment, że w efekcie różnych transformacji dostajemy identyczne obiekty w naszych dataframach. Jest to szczególnie uciążliwe, gdy musimy pracować na unikatowych danych. Poza tym jednak najzwyczajniej w świecie niepotrzebnie zajmują one miejsce. Aby tego uniknąć, usuwamy duplikaty. Tylko jak to zrobić w Sparku? Istnieją dwa sposoby – distinct() oraz dropDuplicates().

Distinct jest najprostszą formą usuwania duplikatów. Tutaj sprawa jest prosta – na dataframe wywołujemy funkcję distinct(), po czym Spark wyszukuje rekordy, które w całości się pokrywają i zostawia tylko jeden z nich. Tak więc w tym przypadku wszystkie kolumny muszą się pokrywać, aby rekord został uznany za duplikat.

Implementacja:

sampleDF.distinct()

Sprawa ma się nieco inaczej, jeśli weźmiemy na warsztat funkcję dropDuplicates(). W tym przypadku możemy wybrać które kolumny bierzemy pod uwagę. W takiej sytuacji “duplikatem” może być np. osoba o tym samym imieniu i nazwisku, ale z innym PESELem.

Implementacja z przykładowymi kolumnami:

sampleDF.dropDuplicates("name", "age")

Co ważne – po takim wywołaniu, zwrócony dataframe oczywiście będzie zawierał wszystkie kolumny, nie tylko wyszczególnione w “dropDuplicates()”.

Tak więc, podsumowując – różnica polega na tym, że distinct() bierze wszystkie kolumny, zaś dropDuplicates() pozwala nam wybrać o które kolumny chodzi.

Explode vs ExplodeOuter

Niekiedy posługując się dataframe, stykamy się z kolumną, która zawiera listę – np. listę imion, liczb itd. Czasami w takiej sytuacji potrzebujemy, żeby te dane znalazły się w osobnych wierszach. Aby to zrobić, używamy funkcji explode – a właściwie jednej z dwóch “eksplodujących” funkcji.

Różnica jest bardzo prosta: explode() pominie nulle, natomiast explode_outer() rozbuduje naszą strukturę także o nulle.

Tak więc, jeśli chcemy rozbić nasze dane pomijając pry tym wartości typu null – wybierzemy explode().

Podsumowanie

Mam nadzieję, że zestawienie powyższych metod rozwieje różne wątpliwości, które mogą niekiedy przychodzić. Spark to złożona, duża technologia. Jeśli chcesz zgłębić ją bardziej – przekonaj swojego szefa, żeby zapisał Ciebie i Twoich kolegów/koleżanki na szkolenie ze Sparka. Omawiamy całą budowę od podstaw, pracujemy na ciekawych danych, a wszystko robimy w miłej, sympatycznej atmosferze;-) – Zajrzyj tutaj!

A jeśli chcesz pozostać z nami w kontakcie – zapisz się na newsletter. Koniecznie, zrób to!

 

Loading
Zrozumieć Sparka: joiny

Zrozumieć Sparka: joiny

Spark to rozległa i złożona technologia. Zrozumienie go wymaga zarówno silnych podstaw teoretycznych, jak i konkretnej, mozolnej praktyki. W serii “Zrozumieć Sparka” postaram się dogłębnie przybliżyć różne aspekty tego zacnego frameworku do przetwarzania danych.

Zaczynamy “z grubej rury” od czegoś, co wycisnęło niejedną łzę u Inżynierów Big Data. Chodzi oczywiście o joiny – mechanizm znany także z tradycyjnego SQL. Z pewnością pojawią się na ten temat także artykuły zaawansowane, dlatego dzisiaj zaczniemy dość lekko – od krótkiego omówienia typów joinów. Wiem, wiem – niby temat znany. W praktyce jednak czasem niedopowiedziany. Niech więc zostanie rozwiązany od podstaw;-).

Czym są joiny w Sparku?

Joiny to mechanizm znany dobrze ludziom pracującym z relacyjnymi bazami danych. Polega na łączeniu tabel. Wyobraź sobie, że masz tabelę, w której są zebrane dane o pracownikach – pesele, imiona, nazwiska, wykształcenie… Zaraz obok możesz zobaczyć tabelkę, w której widać dane dotyczące stanowisk – nazwy stanowisk, pensje, nazwa departamentów w firmie, pesele itd.

Jeśli chcemy pracować na danych dotyczących spraw stanowiskowych w kontekście ludzi – musimy te dwie tabele połączyć. I właśnie temu służą joiny;-). Problematyka w teorii dość prosta, w praktyce ma kilka zagwozdek o których warto wspomnieć. Przede wszystkim właśnie ta podstawowa – czyli różnice w joinach.

Typy joinów w Sparku

Chcąc połączyć tabele, mamy do dyspozycji kilka typów takiego mechanizmu. Podstawowe to inner join, left (outer) join, right (outer) join, full outer join oraz cross join. Dodatkowo możemy w sparku skorzystać z jeszcze dwóch: semi joina oraz anti joina. Omówienie każdego z nich będzie prostsze, jeśli spojrzymy najpierw na ten popularny w internecie diagram.

Wyjaśnijmy czym są owe kółeczka oznaczone jako A oraz B. Jak już napisałem, w tradycyjnych bazach danych będą to tabele. W Sparku będą to Dataframy lub Datasety. Skupimy się tutaj na Dataframach (alias dla Dataset[Row]), gdyż jest to struktura tabelaryczna, stricte przygotowana właśnie pod pracę na tabelach.

Na powyższym obrazku mamy zawsze dwie strony – lewą (A) i prawą (B). Dla wyjaśnienia: to który dataframe jest po lewej a który po prawej to rzecz czysto umowna, w zależności od tego po której stronie jaki dataframe umieścimy w kodzie. Podstawowy schemat tworzenia joina wygląda bowiem następująco:

firstDF.join(secondDF, columns, "joinType")

tak więc w takiej sytuacji “firstDF” będzie “lewym” dataframe, a “secondDF” będzie “prawy”.

Zanim przejdziemy do omówienia poszczególnych typów, zachęcam do przyjrzenia się powyższemu obrazkowi. Jeśli natomiast zapamiętać będzie ciężko, być może poniższe dzieło odrobinę pomoże:

 

W przykładach będę posługiwał się następującymi strukturami:

  1. Dataframe peopleDF zawiera id, firstName, lastName i age.
  2. Dataframe jobsDF zawiera id oraz job – czyli nazwę zawodu (np. “teacher”).

Wyglądają mniej więcej tak:

 

 

Inner Join

Zaczynamy od najprostszego typu, będącego jednocześnie domyślnym podczas łączenia tabel w Sparku. Innej join doprowadza do połączenia tych rekordów, które są wspólne – czyli wtedy, gdy warunek zostanie spełniony.

Ponieważ w Sparku inner join jest domyślny, nie musimy go określać podczas łączenia tabel. Obie poniższe implementacje są poprawne:

val innerJoin: Dataset[Row] = peopleDF.join(jobsDF, "id")
val innerJoin2: Dataset[Row] = peopleDF.join(jobsDF, Seq("id"), "inner")

Generują one następujący efekt:

Left (outer) join

Left join, czy też posługując się pełną nazwą – left outer join, to mechanizm, który łączy obie tabele po konkretnych kolumnach, natomiast zapewnia, że wszystko z lewej strony będzie miało miejsce w finalnej tabeli. Jeśli więc nie uda się powiązać (“zmatchować”) rekordów, kolumny z “prawej strony” będą wypełnione nullami.

Implementacja wygląda następująco:

val leftJoin: Dataset[Row] = peopleDF.join(jobsDF, Seq("id"), "left_outer")

I daje efekt w postaci takiej tabelki:

Zwróć uwagę na to że rekordy w peopleDF kończą się na id 18, natomiast tabela job posiada id od 1 do 13, a potem 19 i 20. Na tym przykładzie widać, że część nie powiązana została wypełniona nullami.

Right (outer) join

W przypadku right joina sprawa jest prosta – to po prostu “lustrzane odbicie” left joina. Łączy on więc obie tabele po konkretnych kolumnach, natomiast zapewnia, że wszystko z prawej strony znajdzie swoje miejsce w finalnej tabeli. Jeśli więc któreś rekordy z prawej tabeli nie będą miały swojego odpowiednika (match nie zostanie zrealizowany/“nie zmatchuje się”), to w tabeli finalnej kolumny z lewej w tych rekordach będą nullami.

Implementacja:

val rightJoin: Dataset[Row] = peopleDF.join(jobsDF, Seq("id"), "right_outer")

Nawet jeśli powyższy opis brzmi jak skomplikowany, poniższy efekt powinien wszystko rozjaśnić:

Full outer  join

Full outer join jest przeciwieństwem inner joina – bierze wszystko. Tak więc można powiedzieć, że jest sklejeniem left i right joinów wraz z usunięciem duplikatów.

Implementacja:

val fullJoin: Dataset[Row] = peopleDF.join(jobsDF, Seq("id"), "full")

Efekt:

Cross join (cartesian/iloczyn kartezjański)

Cross join nazywany jest także iloczynem kartezjańskim. Pochodzi to od matematycznego mechanizmu operacji na macierzach – właśnie iloczynu kartezjańskiego. Dla ciekawskich, z matematyczną definicją oraz kilkoma ćwiczeniami można zapoznać się tutaj.

Dla nas jednak chyba łatwiejsza będzie “definicja zbanalizowana by Maro” – czyli proste “łączymy każdy rekord z każdym”W związku z tym, że każdy z każdym, nie ma tu potrzeby (ani miejsca) by wpisać kolumnę, po której łączymy.

Implementacja:

val crossJoin: Dataset[Row] = peopleDF.crossJoin(jobsDF)

Jak widać, efekt jest znacznie bardziej okazały, niż przy pozostałych typach łączeń. Przedstawiam tylko część, natomiast w tym przypadku rekordów jest aż 270:

Semi join

Pozostały jeszcze dwa typy joinów, zwykle pomijane. Chociaż są znacznie rzadziej spotykane, warto moim zdaniem znać oba, gdyż oba mogą okazać się naprawdę przydatne. Jeśli taka potrzeba zajdzie, po prostu zastosujemy przygotowaną już konstrukcję, zamiast robić “skok przez plecy” z selectem, równaniami, wherami i filtrami.

Pierwszym z nich jest Semi Join. To nic innego jak inner join, który… pomija kolumny z lewej strony. Tak więc po zmatchowaniu rekordów, brane są pod uwagę tylko kolumny z lewej strony.

Implementacja:

val semiJoin: Dataset[Row] = peopleDF.join(jobsDF, Seq("id"), "left_semi")

Efekt:

Anti Join

Ostatnim, najciekawszym moim zdaniem, joinem – jest anti join. Jest on pewnego rodzaju “dopełnieniem” semi-joina. Albo jego przeciwieństwem – zależy jak patrzeć. Chodzi w każdym razie  to, że chociaż kolumny są takie same – a więc jedynie z lewej strony – to… w finalnej tabeli znajdują się tylko rekordy, które nie zostały zmatchowane, nie da się ich powiązać między tabelami.

Implementacja:

val antiJoin: Dataset[Row] = peopleDF.join(jobsDF, Seq("id"), "left_anti")

Efekt:

Implementacja inaczej

Warto zauważyć jedną rzecz: wszędzie powyżej poszedłem na łatwiznę i łączyłem po takiej samej kolumnie. Może się jednak okazać, że nie będzie takich samych kolumn! Cóż wtedy zrobić? Opcji jest jak zawsze kilka. Przede wszystkim dwie (poniższe konstrukcje nie odpowiadają już schematom których trzymaliśmy się przez większość artykułu):

Opcja nr 1: zróbmy tak, żeby były takie same kolumny. Czyli po prostu zmieńmy nazwy kolumn, które mają być matchowane.

val leftJoin: Dataset[Row] = peopleDF.join(jobsDF.withColumnRenamed("personId", "id"),
      Seq("id"), "left_outer")

Opcja nr 2: powiedzmy, które kolumny mają być matchowane. Tak, to zdecydowanie brzmi rozsądniej:

val leftJoin: Dataset[Row] = peopleDF.join(jobsDF, peopleDF("id") === jobsDF("personId"), "left_outer")

To samo można także zrobić za pomocą następującej składni:

val leftJoin: Dataset[Row] = peopleDF.join(jobsDF, peopleDF("id").equalTo(jobsDF("personId")), "left_outer")

Kolejną rzeczą jaką warto rozważyć jest łączenie po kilku tabelach. Możemy to zrobić bardzo prosto:

val leftJoin: Dataset[Row] = peopleDF.join(jobsDF, Seq("id", "pesel"), "left_outer")

Można także budować bardziej skomplikowane warunki joinów:

val leftJoin: Dataset[Row] = peopleDF.join(jobsDF, peopleDF("id").equalTo(jobsDF("personId")).and(peopleDF("age").gt(jobsDF("minimumAge"))), "left_outer")

Niebezpieczeństwa

Skoro joiny są takie proste i przyjemne, to czy można ich używać bez ograniczeń i bez głębszej refleksji? Otóż… jest dokładnie odwrotnie. Napisałem na początku, że mechanizm ten wycisnął niejedną łzę u inżynierów korzystających ze sparka. Oto dlaczego:

Nie będziemy się tutaj rozwodzić nad tym co i jak w detalu. Sprawa jest prosta: join to uruchomienie shufflingu – czyli przerzucania danych między partycjami oraz nodami. To z kolei bardzo zasobożerna sprawa i warto jej unikać. Na ile oczywiście się da, bo najczęściej się po prostu nie da.

W kolejnych artykułach zagłębimy się w kwestie wydajnościowe właśnie. Na ten moment, z racji że są to podstawy, polecam przejrzeć kod pod kątem akcji, które wykonujemy po joinie. Być może warto po połączeniu tabel zrobić bowiem cache(), persist() lub checkpoint(). Gdybyśmy wywoływali dwukrotnie akcje, dwukrotnie będzie zrobiony także join – a po co, jak nie trzeba?;-).

Ciąg dalszy…

Mam szczerą nadzieję, że wiesz już teraz wszystko czego trzeba, aby zacząć przygodę z joinami w Sparku. Oczywiście w kolejnych artykułach zagłębimy się nieco w tym zagadnieniu – zostań z nami przez newsletter!

Jeśli czujesz, że Twoje wiedza na temat Sparka jest nieuporządkowana i frustruje Cię to, to mam dobrą wiadomość. Organizujemy szkolenia dla firm z zakresu… właśnie Apache Spark. Jeśli masz dobrego szefa, podeślij mu kontakt do nas – możemy pojawić się u Ciebie w firmie i pokazać wszystko od 0 do Spark Inżyniera;-). Strona ze szkoleniem dostępna jest tutaj.

Tymczasem zapraszam do naszego newslettera. Zostańmy w kontakcie i… razem budujmy naszą wspólną, polską społeczność Big Data!

 

Loading

 

Zamiana Dataframe w Dataset w Sparku. Taka, która działa.

Zamiana Dataframe w Dataset w Sparku. Taka, która działa.

Największy problem Dataframe w Sparku? Oczywiście – brak jasności, jaka schema jest w aktualnym DFie. W związku z tym często wykorzystuje się Dataset[T], gdzie T to konkretna Case Class. Dzięki temu, jeśli wywołujemy metodę, która zwraca Dataset[T], możemy być przekonani, że wiemy jakim typem operujemy, jaką nasz obiekt ma schemę.

Niestety, tak dobrze jest tylko w teorii. W praktyce Spark pozwala wyjść poza ramy schematu i dodać lub odjąć kolumny, które nie istnieją w case class T. Jest to bardzo mylące i osoby które o tym niewiedzą mogą przeżyć niezłą eksplozję mózgu. Pytanie – czy da się coś z tym zrobić? Jak najbardziej. Chociaż nie oficjalną ścieżką, bo to właśnie ona wprowadza w błąd.

Standardowe castowanie Dataframe do Dataset[T]

Na początku zobaczmy jak to powinno wyglądać według planu “standardowego”. W przykładzie będziemy się posługiwać case classą “Person”, która ma id, name i age. To do niej będziemy “równać” wszystkie Dataframy. Wykorzystamy do tego trzy DFy:

  1. perfectDF – będzie miał dokładnie takie kolumny jak powinniśmy mieć wykorzystując Person.
  2. bigDF – będzie zawierał jedną kolumnę więcej – “lastName”
  3. smallDF – będzie zawierał jedynie kolumny “id,name”. Będzie więc brakowało kolumny “age”

Dodam jeszcze, że cały kod pisany jest w Scali.

Poniżej przedstawiam uproszczone utworzenie pierwszego Dataframe’a. Reszta jest na podobnej zasadzie.

val perfectDF: Dataset[Row] = Seq(
      (1, "John", "26"),
      (2, "Anna", "28")
    ).toDF("id", "name", "age")

Czas na najważniejszą rzecz: oficjalnym i najbardziej popularnym sposobem na zamianę DF w DS[T] jest wywołanie df.as[T]. Aby to zrobić, musimy wcześniej zaimportować implicit._. Można ewentualnie zrobić także df.as[T](Encoders.product[T]).

Po takim “myku” spodziewamy się, że otrzymamy zmienną z typem Dataset[T]. W naszym przypadku, po zastosowaniu perfectDF.as[Person] faktycznie tak się dzieje. Wywołujemy więc “show()” aby sprawdzić jak wyglądają nasze dane i… wszystko gra. Mamy więc Dataset o jasnej schemie, gdzie możemy odwoływać się do konkretnych pól.

Problem zaczyna pojawiać się w sytuacji, gdy nasz Dataframe nie jest idealnie wpasowany do schematu klasy T. Przykładowo – zmienna bigDF to Dataframe, który zwiera kolumny “id”, “name”, “lastName”, “age”. Po wywołaniu metody bigDF.as[Person] możemy się spodziewać, że kolumna zostanie obcięta. Wszak w wyniku operacji dostajemy Dataset[Person], a klasa Person nie zawiera pola “lasName”. Niestety, jest zupełnie inaczej.

val bigDF: Dataset[Row] = Seq(
      (1, "John", "Smith", "27"),
      (3, "Anna", "Smith", "1")
    ).toDF("id", "name", "lastname", "age")

val bigDFAsPerson: Dataset[Person] = bigDF.as[Person]
bigDFAsPerson.show()
bigDFAsPerson.printSchema()

Efekt – w postaci schemy oraz wyglądu tabelki – można obserwować poniżej:

+---+----+--------+---+
| id|name|lastname|age|
+---+----+--------+---+
|  1|John|   Smith| 27|
|  3|Anna|   Smith|  1|
+---+----+--------+---+

root
 |-- id: integer (nullable = false)
 |-- name: string (nullable = true)
 |-- lastname: string (nullable = true)
 |-- age: string (nullable = true)

Sprawa ma się inaczej, gdy spróbujemy dokonać konwersji do klasy z Dataframe, który ma zbyt mało kolumn:

val smallDF: Dataset[Row] = Seq(
  (1, "maro"),
  (3, "ignacy")
).toDF("id", "name")

val smallDFAsPerson: Dataset[Person] = smallDF.as[Person]

smallDFAsPerson.show()
smallDF.printSchema()

Efekt: ponieważ nie ma jednej z kolumn, Spark rzuci wyjątek. A konkretniej AnalysisException.

21/08/30 21:25:11 INFO CodeGenerator: Code generated in 25.8197 ms
Exception in thread "main" org.apache.spark.sql.AnalysisException: cannot resolve '`age`' given input columns: [id, name];
    at org.apache.spark.sql.catalyst.analysis.package$AnalysisErrorAt.failAnalysis(package.scala:42)
    at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$checkAnalysis$1$$anonfun$apply$3.applyOrElse(CheckAnalysis.scala:110)
    at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$checkAnalysis$1$$anonfun$apply$3.applyOrElse(CheckAnalysis.scala:107)
    at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$transformUp$1.apply(TreeNode.scala:278)
    at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$transformUp$1.apply(TreeNode.scala:278)
    at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:70)
    at org.apache.spark.sql.catalyst.trees.TreeNode.transformUp(TreeNode.scala:277)  
.
.
.
    at App$.main(App.scala:44)
    at App.main(App.scala)

 

Sprawdzony sposób na zamianę Dataframe do Dataset[T]

Na szczęście – jest możliwość, aby to naprawić. Aby nie przedłużać powiem tylko, że należy po prostu napisać swój własny kod, dzięki któremu rozszerzymy nieco Dataframe. Najpierw jednak rozpiszmy sobie jakie chcemy spełnić warunki.

Po konwersji z Dataframe, nowo utworzony Dataset:

  1. Nie będzie zawierał niepotrzebnych kolumn.
  2. Będzie miał nowo utworzone kolumny. Będą one wypełnione nullami.

Aby to zrobić, utwórz object SparkExtensions, a następnie dodaj kod:

import scala.reflect.runtime.universe.TypeTag
import org.apache.spark.sql.{DataFrame, Dataset, Encoders, Row}

object SparkExtensions {
  implicit class ExtendedDataFrame(df: DataFrame) {
    def to[T <: Product: TypeTag]: Dataset[T] = {
      import df.sparkSession.implicits._
      import org.apache.spark.sql.functions.col
      val columnsUniqueInT = Encoders.product[T].schema.diff(df.schema)
      val withAdditionalColumns: Dataset[Row] = columnsUniqueInT .foldLeft(df)((previousDF, column) =>
        previousDF.withColumn(column.name,
          lit(null).cast(column.dataType)))
      withAdditionalColumns.select(Encoders.product[T]
        .schema
        .map(f => col(f.name)): _*)
        .as[T]
    }
  }
}

Co tu się dzieje? Przede wszystkim trzy rzeczy:

  1.  Znajdujemy unikalne kolumny – robimy to poprzez różnicę tego co jest w naszej case class oraz w schemacie dataframe.
  2. Tworzymy Dataframe z dodatkowymi kolumnami, które wypełniamy nullami i castujemy do odpowiednich typów
  3. Tworzymy Dataframe (na bazie [2]), w którym wybieramy jedynie te kolumny, które są przewidziane w klasie T.

Jak tego użyć? Musimy zrobić dwie rzeczy: zaimportować nasz object SparkExtensions (import functions.SparkExtensions._), a następnie przy Dataframe zamiast “.as[Person]” zrobić “.to[Person]”.

val bigDFToPerson: Dataset[Person] = bigDF.to[Person]
bigDFToPerson.show()
bigDFToPerson.printSchema()

Efekt:

+---+----+---+
| id|name|age|
+---+----+---+
|  1|John| 27|
|  3|Anna|  1|
+---+----+---+

root
 |-- id: integer (nullable = false)
 |-- name: string (nullable = true)
 |-- age: string (nullable = true)

 

UWAGA! Taka opcja zakłada, że wszystko musi zgadzać się w 100%.

Jak wiadomo, pole w Datasetach sparkowych składa się z 3 rzeczy:

  1. Nazwy
  2. Typu
  3. Określenia, czy jest nullable czy nie – czyli czy może przyjmować null jako wartość.

Dla tych, którzy wolą nieco bardziej liberalną wersję, przygotowałem też metodę, która pozwala okrajać nie uwzględniając trzeciej rzeczy.

def to[T <: Product: TypeTag]: Dataset[T] = {
      import df.sparkSession.implicits._
      import org.apache.spark.sql.functions.col
      val columnsFromT = Encoders.product[T].schema.map(c=> (c.name,c.dataType))
      val columnsFromDF = df.schema.map(c=> (c.name, c.dataType))
      val columnsUniqueInT = columnsFromT.diff(columnsFromDF)
      val withAdditionalColumns: Dataset[Row] = columnsUniqueInT.foldLeft(df)((previousDF, column) =>
        previousDF.withColumn(column._1,
          lit(null).cast(column._2)))
      withAdditionalColumns.select(Encoders.product[T]
        .schema
        .map(f => col(f.name)): _*)
        .as[T]
    }

Zdaję sobie sprawę, że robi się już nieco mało estetycznie, ale zapewniam – działa;-). Stosuje się dokładnie tak samo jak w przypadku poprzedniej wersji.

Liczę głęboko, że taka metoda się przyda – w projekcie w którym pracowałem, problem typów był naprawdę mocno odczuwalny. Cóż – po więcej, choć zwykle nieco bardziej “wysokopoziomowych” artykułów – zapraszam do newslettera;-). Powodzenia!

 

Loading
Najczęstsze problemy ze starym Sparkiem (1.X) – co warto wiedzieć, zaczynając projekt

Najczęstsze problemy ze starym Sparkiem (1.X) – co warto wiedzieć, zaczynając projekt

Każdy kto zmuszony został korzystać ze starej wersji Sparka (<2.0) wie jak wiele wody musi upłynąć, aby zrobić niekiedy – wydawałoby się – proste rzeczy. Liczę, że poniższe zestawienie pomoże w rozwiązaniu choć części problemów.

Spark 1.X (1.6) – czyli przygotuj się na nowe (stare?) doznania.

Właśnie dowiedziałeś się, że znalazłeś się w nowym projekcie. Oczywiście musi z Twojej strony paść pierwsze, zapobiegawcze pytanie: “w jakich technologiach będę robić?”. Odpowiedź która przychodzi sprawia, że rozpływasz się niczym aktorka w reklamie czekolady: “Spark, będzie Spark…”.

Znakomicie! Szkoda tylko, że nikt nie wspomniał co to za Spark. Chodzi bowiem o starego dobrego Sparka 1.6, weterana Big Data. Zasłużony dla branży, odznaczony tysiącem orderów procesingu, choć czasem wolelibyśmy podziwiać go jedynie na starych fotografiach.

Czy jednak Spark <2.0 jest taki zły? Z mojego doświadczenia powiem dość prosto: jeśli przestawisz się i zdobędziesz zawczasu odpowiednią wiedzę, unikniesz gorzkich rozczarowań. Nie są to jednak rzeczy, które drastycznie zmieniają posługiwanie się Apache Spark i ciągle jest to solidna, przemyślana technologia do zrównoleglenia obliczeń. Jeśli pracowałeś/aś wcześniej głównie na 2.X (o “trójce” nie wspominam), to mam dobrą wiadomość: nie przeżyjesz szoku. Co najwyżej od czasu do czasu zmarszczysz czoło i sapniesz pod nosem to i owo. Na szczęście twórcy Sparka bardzo dobrze zadbali o kompatybilność wsteczną. Pracując tu, nie musimy zmagać się z koszmarami znanymi z innych technologii (pozdrawiam wszystkich programistów Scala Play – trzymajcie się tam!).

Poniższa lista jest moim (a także kilku osób które poprosiłem o radę – dzięki wielkie Panowie) subiektywnym zbiorem informacji, które spowodują, że praca w nowym projekcie ze “starym sprzętem” będzie naprawdę fajna. Mam przynajmniej taką nadzieję:-).

Subiektywna lista zmian w starym Sparku, które warto znać

Zacznijmy od spisania wszystkich rzeczy. Potem niektóre z nich (te, które nie są oczywiste zaraz po przywołaniu do tablicy;-)) opiszę nieco dokładniej.

  1. Brak możliwości wczytywania z CSV do DataFrame.
  2. Inna praca ze SparkContext / SparkSession
  3. Spark 1.X działa ze Scalą 2.10 (Spark 2.X – 2.11, Spark 3.X – 12) – oczywiście teoretycznie podział jest bardziej płynny, ale bardzo możliwe, że pracując ze Sparkiem 1.X, zmuszony/a będziesz robić to we współpracy ze starszą wersją Scali.
  4. Rozwoj Machine Learning od Sparka 2.0
  5. Spark Structured Streaming zostaje wprowadzony dopiero od Sparka 2.0.
  6. Od wersji 2.0 Dataframe staje się aliasem dla Dataset[Row], obie kolekcje mają ujednolicony interfejs.
  7. Poniżej 2.0 gdzie indziej znajdywały się encodery.
  8. Po prostu… wydajność. W codziennej, inżynierskiej robocie tego nie widzimy, ale przy procesowaniu to ma znaczenie.

Brak możliwości wczytywania z CSV do Dataframe

Jak wczytać dane z CSV? To proste!

val sampleDF = spark.read
      .option("header","true")
      .csv("sample_data.csv")

No jednak… nie. To jest proste, jeśli korzystasz ze Sparka >=2.0. Niestety, w “jedynkach” sprawa miała się nieco gorzej. Nie dysponujemy po prostu odpowiednią biblioteką, która dałaby nam dostęp do źródła typu csv. Na szczęście łatwo to naprawić.

Przede wszystkim – otwieramy nasz build.sbt (jeśli używamy sbt, inaczej maven itd.) i dodajemy odpowiednią bibliotekę:

libraryDependencies ++= Seq(
  ...
  "com.databricks" % "spark-csv_2.10" % "1.5.0",
  ...
)

Następnie, już podczas wczytywania danych:

sqlContext.read
      .format("com.databricks.spark.csv")
      .option("header", "true")
      .load(sample_data.csv)

I już:-). Zauważ tylko, że wykorzystujemy tu sqlContext, a nie sparkSession.

Inna praca ze SparkContext / SparkSession

A właściwie to… po prostu brak SparkSession. Został dodany od Sparka 2.0, więc jeśli używasz nieco bardziej leciwych wersji – zapomnij, że coś takiego istnieje. Zamiast tego musisz nauczyć się żonglować SparkContext oraz SQLContext.

Jak to wygląda w praktyce? Na przykład tak:

val conf = new SparkConf().setAppName("RDFSparkAppName")

val sc: SparkContext = new SparkContext(conf)
val sqlContext: SQLContext = new org.apache.spark.sql.SQLContext(sc)

// UDFs REGISTERING
val pseudonymisationUDF: PseudonymisationUDF = new PseudonymisationUDF
sqlContext.udf.register("pseudonymisationUDF", pseudonymisationUDF, DataTypes.StringType)

// DATAFRAME CREATING val sampleDF: DataFrame = sqlContext.read .format("com.databricks.spark.csv") .option("header", "true") .load("sample_data.csv")
// RDD CREATING val sampleRDD: RDD[Array[Byte]] = sc.parallelize(ids) .map(i=> Bytes.toBytes(i))

W większości będzie trzeba operować na sqlContext, ale jak widać zdarzają się sytuacje, gdy trzeba użyć także sparkContext. Warto wiedzieć jednak, że sparkContext da się wyciągnąć z sqlContext.

Rozwój Machine Learning od Spark 2.0

I tu istotna zmiana: popularny moduł Spark MlLib, oparty o RDD, od sparka 2.0 przechodzi w stan utrzymania (maintanance). Choć w dokumentacji sparka 2.X możemy przeczytać, że prawdopodobnie od Sparka 3 MlLib będzie wyłączony – tak się nie stało. Ciągle możemy go używać, jednak nie jest już rozwijany.

W zamian jednak, od Sparka 2.0 mamy do dyspozycji Spark ML. To moduł oparty o Dataframe. Co ciekawe – Spark Ml nie jest oficjalną nazwą, natomiast jest to pakiet, w którym znajdziemy interesujące nas klasy.

Zmiany w Dataset i Dataframe

Oczywiście nie można nie wspomnieć o najważniejszej zmianie. Jeśli korzystasz ze Sparka <2, najprawdopodobniej będziesz używać RDD. Jest jednak opcja, żeby pracowć z Dataframe. Należy jednak pamiętać, że największa magia optymalizacyjna zaczyna się w tym temacie dopiero od Spark 2.0. Od tego momentu Dataframe staje się aliasem dla Dataset[Row], zaś Dataset i Dataframe mają ujednolicony interfejs.

Encodery w innym miejscu w Sparku 1.X

Encodery wykorzystujemy w Sparku aby przekonwertować DataFrame a Dataset. Robimy to za pomocą metody “as[T]”, gdzie “T” to konkretna case class. Co prawda nie jest to najlepszy sposób na otrzymanie silnego, klarownego Dataset, ale jest oficjalny, dlatego go tutaj przedstawiam.

Spark 1.X

case class Person(id: Int, name: String)

val encoder = org.apache.spark.sql.catalyst.encoders.ExpressionEncoder[Person]

// creating some DF
val someDF = ...

val dataset = someDF.as(encoder)

Spark >= 2.X

case class Person(id: Int, name: String)
val encoder = org.apache.spark.sql.Encoders.product[Person]

// creating some DF val someDF = ...

val dataset = someDF.as(encoder)

Wydajność

Pisząc kod oczywiście nie mamy styczności z wydajnością. Nie zmienia to jednak postaci rzeczy, że podczas wykonywania joba to właśnie ona ma kluczowe znaczenie. I tutaj po prostu warto pamiętać, że pierwsze Sparki były znacznie, znacznie wolniejsze i posiadały mniejsze możliwości.

W tym eksperymencie porównywano Sparka 1.6 oraz 2.0. Sumowano miliard liczb, dokonywano joina oraz sprawdzano groupBy. Największe wrażenie robią dwa pierwsze testy – wyniki w tabelce.

 

Primitive

Spark 1.6

Spark 2.0

Sum

15.6
seconds

0.68
seconds

Join

58.12  seconds

0.98
seconds

Warto oczywiście pamiętać, że taki eksperyment niesie szereg wątpliwości (np. kwestia nodów). Nie zmienia to jednak postaci rzeczy, że między sparkiem 1.6 a 2.0 jest spora różnica. Wynika ona z optymalizacji, jakie zostały przeprowadzone pod spodem Datasetów oraz Dataframów. Co warto wiedzieć – między ostatnią wersją 2.X oraz 3.0 także wprowadzono optymalizacje, dzięki którym możliwości Sparka stały się jeszcze większe. Twórcy chwalą się nawet, że 3 jest nawet 17x szybszy, co zostało wykazane w teście TPCDS.

Podsumowanie

Jeśli już siedzisz w projekcie ze Sparkiem 1.X – nie masz wyjścia, trzeba robić. Pamiętaj, że nie jest taki zły, a pod kątem inżynierskim jest nawet dość mocno podobny. Bez obaw, po prostu postaraj się pamiętać o tym jakie rzeczy mogą Cię zaskoczyć i… rób swoje. Powodzenia!

A – właśnie. Jeśli jesteś zainteresowany/a tego typu rzeczami, zapisz się na newsletter RDF. Wspólnie tworzymy polską społeczność Big Data. Liczę, że od dzisiaj także z Tobą;-).

 

Loading