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:
- Dataframe peopleDF zawiera id, firstName, lastName i age.
- 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!