(py)Spark, Hadoop i HDFS – podstawy

Czytanie zajmie Ci około 9 minut

Dzisiaj zajmiemy się wykorzystaniem Sparka i Hadoopa do przetwarzania większej ilości danych. Oraz do budowania prostego modelu (regresji liniowej). Może jeszcze nie jest to big data, ale mechanizmy są identyczne jak w przypadku większej liczby danych. Wystarczy tylko tych danych więcej zgromadzić, zbudować większe środowisko (dużo serwerów) i… też będzie działało.

A w dodatku poznamy pakiet faker który pozwoli nam na wygenerowanie sztucznych danych.

Spark i Hadoop

Apache Spark to rozwijana na zasadach open source platforma klastrowego przetwarzania danych, która ma interfejsy API dla takich języków programowania jak Scala, Python, Java i R.

Apache Hadoop to z kolei skalowalna platforma do przechowywania i zarządzania dużymi zbiorami danych. Ogólna koncepcja polega na podziale dużych zbiorów na mniejsze fragmenty, które mogą być równolegle przetwarzane w wielu węzłach wykorzystujących standardowe serwery. Oprócz tego Hadoop obsługuje dane niestrukturalne i jest niezależny od systemu operacyjnego. My wykorzystamy przede wszystkim HDFS (Hadoop Distributed File System) czyli system plików zapewniający wysoką wydajność dostępu do danych.

To taki tekst dla inżynierów big dejta. Dla analityków i programistów: Sparkiem będziemy sięgać do danych leżących w wielu plikach na HDFS i traktować je jako jedną tabelkę. To jest ta cała magia która jest najfajniejsza. Dodatkowo – wielkość tych danych w plikach (na Hadoopie/HDFSie) nie ma za bardzo znaczenia – pamięci nie zabraknie (bo jakoś Spark sobie z tym poradzi). No chyba, że dane ze Sparka zechcemy do tej pamięci pobrać (np. do data frame w pandas).

Nie będę opisywał jak zainstalować Sparka i Hadoopa. Jest do tego świetny tutorial, który przeszedłem krok po kroku i u mnie działa. Co więcej – nawet instalowałem wersje nowsze niż w tym tutorialu (odpowiednio trzeba pozmieniać linki do pobieranych instalek i później same nazwy plików czy folderów. Ale to proste jest. No i oczywiście tutorial jest przeznaczony na Linuxa, bo takie rzeczy to nie na Windows (chociaż pewnie się da).

Tutorial tutaj: Installing and Running Hadoop and Spark on Ubuntu 18

Podstawy Sparka

Na początek kilka podstaw – jak utworzyć tabelkę (data frame) w Sparku? Jak wczytać plik? Jak wybrać kolumnę? Jak odfiltrować dane? Jak je posortować i inne takie. Zobaczmy na kilka prostych przykładów:

kraj miasto liczba
PolskaWarszawa1
PolskaKraków2
NiemcyBerlin1
NiemcyHamburg3
CzechyPraga1
RosjaMoskwa1
FrancjaLyon4

Prawda, że jak na razie proste? To tylko króciutkie wprowadzenie, a nawet zarys takiego wprowadzenia, więc o szczegółach musicie doczytać. Jedźmy dalej:

krajmiasto
PolskaWarszawa
PolskaKraków
NiemcyBerlin
NiemcyHamburg
CzechyPraga
RosjaMoskwa
FrancjaLyon
krajmiastoliczba
PolskaKraków2
NiemcyHamburg3
FrancjaLyon4
krajmiastoliczba
NiemcyHamburg3
FrancjaLyon4
krajmiastoliczba
CzechyPraga1
FrancjaLyon4
NiemcyBerlin1
NiemcyHamburg3
PolskaKraków2
PolskaWarszawa1
RosjaMoskwa1
krajmiastoliczba
CzechyPraga1
FrancjaLyon4
NiemcyHamburg3
NiemcyBerlin1
PolskaKraków2
PolskaWarszawa1
RosjaMoskwa1
kraj avg(liczba)
Rosja1.0
Francja4.0
Czechy1.0
Niemcy2.0
Polska1.5

A co jeśli chcemy kilka agregatów na raz, na tej samej kolumnie i od razu zmieniając nazwę kolumny wynikowej?

kraj minimum maximum srednia suma
Rosja111.01
Francja444.04
Czechy111.01
Niemcy132.04
Polska121.53

Znak \ pozwala na złamanie linii co ułatwia czytanie i pisanie kodu. Nie jest to niestety R-owy pipe %>%, ale zawsze coś.

kraj miasto liczba nowe_miasto
PolskaWarszawa1Warszawa
PolskaKraków2Kraków
NiemcyBerlin1Berlin
NiemcyHamburg3Hamburg
CzechyPraga1Praga
RosjaMoskwa1Moskwa
FrancjaLyon4Lyon
kraj miasto liczba liczba_x1000
PolskaWarszawa11000
PolskaKraków22000
NiemcyBerlin11000
NiemcyHamburg33000
CzechyPraga11000
RosjaMoskwa11000
FrancjaLyon44000

Niektórzy wolą zapytania SQL-owe i dla takich ludzi Spark to umożliwia. Ot, po prostu, robimy widok SQL-owy na tabelę w Sparku, a później operujemy na danych używając zapytań SQL:

krajmiasto liczba
PolskaWarszawa1
PolskaKraków2
NiemcyBerlin1
NiemcyHamburg3
CzechyPraga1
RosjaMoskwa1
FrancjaLyon4
kraj liczba_miast
Rosja1
Francja1
Czechy1
Niemcy2
Polska2

SQLa warto się nauczyć chociaż na podstawowym poziomie. Na pewno będzie sprawdzany na przeróżnych rekrutacjach, ale też w pracy z danymi z przeróżnych baz danych się przydaje.

Po wykonaniu wszelakich agregacji, wyliczeń i innych manipulacji danymi możemy je pobrać prosto ze Sparka do Padnas i cokolwiek z tym dalej potrzeba robić. Wykresy rysować czy coś.

kraj miasto liczba
0PolskaWarszawa1
1PolskaKraków2
2NiemcyBerlin1
3NiemcyHamburg3
4CzechyPraga1
5RosjaMoskwa1
6FrancjaLyon4

Dane fejkowe

Zanim zaczniemy sensowniej wykorzystywać Sparka przygotujemy sobie trochę fejkowych danych. Wykorzystamy do tego pakiet Faker. Niech będą to osoby, które będą opisane przez imię, nazwisko, płeć i wiek, a także adres domowy i adres pracy. Dodatkowo dla pracy dodamy stanowisko i wynagrodzenie. Struktura dla każdej z osób będzie następująca:

Jak widzicie wygląda to jak struktura pliku JSON. I bardzo dobrze, bo będziemy generować takie właśnie pliczki. Tutaj jeszcze umyślnie jest tak, że plik ma zagnieżdżoną strukturę co w przypadku na przykład wczytania do Pandas może powodować problemy (szczególnie jak czasem dany element jest czasem go nie ma).

Wszystkich tych danych nie wykorzystamy w naszym przykładzie, ale generator umożliwi Wam przygotowanie różnych zestawień w różnych przekrojach – ile średnio zarabia kobieta na określonym stanowisku? Możecie sobie poćwiczyć sparkowe zapytania, ale uprzedzam – bez sensownych danych trudno zweryfikować czy wychodzi sensownie. Dodatkowo Faker całkiem nieźle losuje dane, więc może być jeszcze trudniej (dane są dobrze losowo rozłożone).

Cały skrypt poniżej, mam nadzieję że komentarze wyjaśniają co i jak:

Powyższy kod wygeneruje nam 2000 plików JSON z losowymi danymi wymyślonych osób.

Zwróćcie uwagę, że wynagrodzenie zależy liniowo od wieku osoby (jest około 200 razy większe, plus minus kilka setek). To będziemy estymować w naszym modelu.

Mając tak przygotowane pliki możemy spróbować je wczytać do data frame w Sparku. Najpierw jeden plik:

dom imie nazwisko plec praca wiek
[01-516, Przemyśl…AgnieszkaRejniakK[FPUH Fitrzyk, 96…35

Zwróćcie uwagę, że kolumny dom i praca to listy. A gdy sprawdzimy schemat jaki Spark sam rozpoznał?

Ładnie widać wszystkie zagnieżdżenia, a nawet typy danych które zawsze są numeryczne (wiek i wynagrodzenie).

Sprawdźmy czy możemy wczytać wszystkie 2000 JSONów na raz?

dom imie nazwisko plec praca wiek
[80-540, Łuków, 8…UrszulaSkałaK[Spółdzielnia Dyb…53
[86-924, Tomaszów…MaurycyPszczółkaM[Gabinety Forysia…49
[59-091, Mińsk Ma…TadeuszMikusM[Fundacja Choroś-…35
[31-706, Wągrowie…KlaraBartłomiejczykK[Spółdzielnia Pos…65
[58-611, Świnoujś…JuliannaKamyszekK[FPUH Rząca-Jama …46
[34-534, Bytom, 9…SylwiaOpasK[Stowarzyszenie L…62
[97-513, Łuków, 3…NatanielSłykM[Stowarzyszenie H…49
[67-145, Piekary …OlgierdKorkoszM[Spółdzielnia Rom…33
[97-413, Bielsk P…MelaniaKrekoraK[Spółdzielnia Chw…57
[51-127, Siemiano…DawidNasiadkaM[Stowarzyszenie P…61
[73-637, Żywiec, …AdriannaNowotnikK[Fundacja Froń-Fa…30
[13-696, Wyszków,…MalwinaLeżońK[Stowarzyszenie P…57
[69-089, Głogów, …AnastazjaPotoczekK[Gabinety Jaje-Ge…54
[98-298, Bolesław…RozaliaKapustkaK[Stowarzyszenie W…43
[54-050, Marki, 7…PrzemysławGorzelańczykM[Stowarzyszenie F…53
[68-995, Sandomie…TadeuszOleksikM[Watras-Bogaczyk …41
[41-103, Otwock, …KrystianJażdżykM[Gabinety Becker-…29
[04-831, Jarosław…TomaszSmektałaM[Fundacja Morgała…37
[31-336, Gdańsk, …OlafCzuryłoM[Spółdzielnia Czu…58
[76-335, Wągrowie…DominikDrwięgaM[Spółdzielnia Mie…37

To tylko początek tabeli, wszystkich wierszy mamy:

No dobrze Panie Bobrze, ale jak wyciągnąć te wartości w zagnieżdżonych kolumnach? Dom i praca mają swój kod pocztowy – wyjmijmy te wartości:

kod_dom kod_praca
80-54002-055
86-92451-564
59-09183-771
31-70683-631
58-61150-091

Zapisanie na HDFS (skopiowanie plików z konsoli)

Te wygenerowane pliki dla każdej z osób chcemy mieć na Hadoopie (na HDFS). Tak, żeby sięgnąć do nich Sparkiem. Oczywiście Sparkiem można sięgnąć i do plików z dowolnego katalogu (jak w przypadku tworzenia files_df), ale my chcemy z HDFS (do celów edukacyjnych).

Proste to zadanie, z konsoli wykonujemy po prostu:

Jak już wrzucimy pliki na HDFS to jak je wczytać w Sparku? A to za chwilę, w ramach kolejnych kroków.

Regresja liniowa

W Sparku cechy (czyli wartości X) muszą być jedną kolumną zawierającą wektor. Odpowiedź (wartość Y) może być już jedną kolumną, nie trzeba jej specjalnie przekształcać.

Takie przekształcenie zapewni nam VectorAssembler. Model będzie liniową regresją, więc od razu zaimportujmy stosowne funkcje z pySparka:

features wynagrodzenie
[53.0]10500
[49.0]9800
[65.0]13300
[35.0]6700
[46.0]9000

Nasz X czyli features jest wektorem (jednym na każdy wiersz, jest to przy okazji wektor jednoelementowy), a Y – po prostu są to wartości. Zdefiniujmy teraz model i wytrenujmy go:

Jakie mamy współczynniki naszej prostej wychodzącej z modelu?

Nachylenie prostej to okolice 200. Pamiętacie jak wyglądało generowanie danych? Wynagrodzenie to wiek razy 200 plus minus jakieś wartości. Z tego plus minus zrobił się ułamek przy 200.

Sprawdźmy jeszcze współczynniki mówiące o ocenie modelu – w tym przypadku błąd średnio kwadratowy (RMSE) i R2 na danych treningowych.

Oczywiście możemy zrobić też predykcję – po co model jak nie można nic przewidzieć?

prediction wynagrodzenie features
5802.7566335915125600[29.0]
7405.00243766381857400[37.0]
9207.5289672451649000[46.0]
9808.371143772289800[49.0]
10008.65186928131710200[50.0]
13012.86275191689512500[65.0]
5402.1951825734355200[27.0]
6203.3180846095886600[31.0]
7605.2831631728587100[38.0]
8406.406065209018200[42.0]

I sprawdzamy R2 oraz RMSE na danych testowych:

Wyniki są zbliżone do danych treningowych (nie dziwne to w tym przypadku), myślę że dość zadowalające (chociaż mieć prawie 315 złotych a nie mieć 315 złotych to jednak 630 zł różnicy…).

Model w scikit-learn

A jak podejść standardowo do tych samych danych i przygotować dla porównania model z użyciem np. scikit-learn? Trzeba dane wyjąć ze Sparka do Pandas, a dalej już standardowo. Zatem dla przypomnienia:

wynagrodzenie wiek
1050053
980049
1330065
670035
900046

Widzimy, że współczynnik nachylenia prostej też jest w okolicach 200, R2 podobnie jak poprzednio bardzo blisko jedynki, błąd średnio kwadratowy w okolicach 315 – czyli też jak poprzednio. Wyniki mamy więc dość podobne.

Jaka jest różnica? I po co to wszystko? Ano po to, że Spark poradzi sobie z bardzo dużymi danymi, Pandas i SciKit-Learn może braknąć pamięci w którymś momencie. Dodatkowo Spark potrafi rozproszyć obliczenia na wszystkie maszyny w klastrze, a scikit policzy lokalnie, na jednej maszynie. Oczywiście mając tylko jedną maszynę to wielkiego przyspieszenia nie zauważymy (a nawet obliczenia będą sumarycznie dłuższe – Spark musi wystartować co swoje też zajmuje). Są jednak przykłady zadań, gdzie dane obrabiane Sparkiem (wtedy bodaj bez HDFS) umożliwiły wykonanie zadania. Wówczas próbowałem w R, bo Sparka w R też można użyć.

2 myśli na “(py)Spark, Hadoop i HDFS – podstawy”

  1. Świetny artykuł. Dobrze się czyta, z niecierpliwością czekam na następne części. Może przetwarzanie strumieniowe on-line Spark / Flink?

    1. Kolega z Wiadrodanych.pl więcej o tym pisze, polecam. Chyba nie ma sensu się powtarzać… ale zawsze jakiś pomysł się może trafić. Tutaj chciałem wprowadzić temat i pokazać z czym to się je.

Dodaj komentarz

Twój adres email nie zostanie opublikowany. Pola, których wypełnienie jest wymagane, są oznaczone symbolem *