Spark to pandas na sterydach

Przeczytaj ten artykuł, jeśli zakochałeś się kiedyś w bibliotece pandas i zacząłeś się zastanawiać, czy może równoległe przetwarzanie danych da się ogarnąć.

Kim jesteś?

Niech zgadnę. Jesteś Data Scientist - przetwarzasz dane zawodowo, używasz ich do budowania modeli, a wszystko to z użyciem pandas i innych superbibliotek. A może po prostu jesteś dojrzałym programistą, który kiedyś usłyszał o przetwarzaniu równoległym, ale nigdy nie było okazji zobaczyć na własne oczy przypadku użycia, który by to uzasadniał. Nawet jeśli nie należysz do żadnej z tych grup, zastanów się: 

Dlaczego potrzebujemy równoległego przetwarzania danych? 

Dlaczego powinno Cię to zainteresować, nawet jeśli wcześniej nie było to Twoją broszką? Opowiem Ci krótką historię... Ostatnio mieliśmy pipeline'a, w którym używaliśmy fbprophet do przewidywania kilku tysięcy serii danych. Cały proces zajmował długie godziny. Wtedy naszła nas myśl: dlaczego nie użyć szybkiego przetwarzania w usługach dostępnych na Azure? Przenieśliśmy kod tam, ale czas wykonania kodu skrócił się tylko kilkukrotnie, mimo, że użyłem tam najlepszych maszyn. Znowu godziny czekania... Gdzie jest wąskie gardło? fbprophet i wszystko to, co było dookoła, było rozwiązaniem na jedną maszynę, którego po prostu nie da się rozproszyć na wiele procesów w sposób naturalny. A jest to coś, czego bym tu oczekiwał: biorę 10 tysięcy wierszy i uruchamiam to na maszynie nr 1, kolejne 10 tysięcy na maszynie nr 2 i tak dalej... Chcesz zobaczyć, jak to zrobić? 

Odpowiedź to Spark 

Apache Spark powstał po to, żeby zindustrializować przetwarzanie równoległe i udostępnić to dla mas. A co było przedtem? Nie tak dawno to Hadoop (ekosystem stworzony dla big data) szalał na salonach. Może dziś niektórzy eksperci twierdzą, że jest już przestarzały. Jednak jedno jest pewne - to był duży krok dla ludzkości. Dzięki Hadoop możesz podzielić swoje duże, obszerne zadanie na mniejsze etapy i rozdzielić je do przetwarzania na kilku maszynach naraz. Potem wynik jest zbierany i voila. 

W tym rozwiązaniu możesz wykorzystać zwykłe serwery, bez inwestowania w topowe maszyny, które też odpowiednio kosztują. Możesz je skalować horyzontalnie (zamiast skalowania wertykalnego). Hadoop jednak nie jest pozbawiony ograniczeń. Musisz użyć technologii MapReduce i języka Java jako jedynego API tego systemu. Poza tym, każdy z kroków przetwarzania uderza w dysk (zapis i odczyt), co jest z pewnością jego wąskim gardłem. 

Właśnie dlatego stworzono Sparka. Jego podstawową zaletą jest przetwarzanie danych w pamięci, co jest dużym krokiem naprzód w porównania do Hadoop. I nie musisz być ograniczony do maszyny, która zarządza całym procesem (driver). Możesz stworzyć tak wiele maszyn wykonawczych (workers), ile jest Ci potrzebne. Wszystko to wspólnoe tworzy klaster (cluster), który jest potężnym narzędziem w Twoich rękach. Nie jesteś sklejony z technologią MapReduce i językiem Java, co dodatkowo może Ci się spodobać. Do dyspozycji masz elastyczne API DataFrame, ze wsparciem dla wielu języków programowania: Pythona, Scali, Javy, R a nawet SQL-a. Brzmi jak coś, co będzie drogie i trudno dostępne? Bynajmniej. Skąd to się bierze? 

Spark czy Spark DIY 

Możesz zainstalować Sparka na własnym komputerze. Ale po co, jeśli mógłbyś wypróbować to narzędzie za darmo online? Nie trzeba instalować bibliotek, martwić się o zależności, żeby wypróbować Sparka. Nawet nie musisz się kłopotać ściąganiem obrazów Dockera. Dzisiejsza technologia charakteryzuje się bogactwem (a może przekleństwem?) PaaS i SaaS. Gotowy na wypróbowanie Sparka? 

Databricks 

Pozwól, że przedstawię Ci Databricks. To produkt, który jest całkowicie online, stworzony dla szeroko pojętej analizy danych. Jest on hostowany na Azure (produkt chmurowy Microsoftu). Właśnie opcja od Azure jest użyta tu jako przykład, ponieważ było to pierwsze miejsce, w którym zetknąłem się z Databricks. 

Tylko na potrzeby tego artykułu stworzyłem sobie konto na darmowej wersji Databricks (community), żeby zobaczyć jakie to jest proste. I tak, jest to proste. W ciągu 2 minut miałem gotowe konto na Databricks, bez wpisywania żadnych danych z kart kredytowych, czy innych zniechęcających elementów. Wystarczy po prostu wypełnić formularz tutaj. Kiedy Twój klaster będzie gotowy, możesz zacząć korzystać z API Sparka. 

Przypadek użycia 

Nie lubię się uczyć czysto teoretycznie. Lubię ten moment, kiedy zauważam, jak jakieś narzędzie przyspiesza rozwiązanie problemu. Może, jako człowiek z branży IT, miałeś kiedyś wątpliwą przyjemność przeglądania długich nudnych kilobajtów logów, żeby zobaczyć, co się stało z aplikacją? Masz kilo-, mega- a może i giga- bajty danych, w których znajdują się pewne wzorce. Możesz nadal to grepować, ale jak wykorzystać tu Sparka? 

Nie musisz (chociaż da się to zrobić) ładować własnych danych, żeby przećwiczyć ten przypadek użycia. Twórcy Databricks umieścili na platformie wiele przykładowych zbiorów danych, które zawierają też właśnie przykładowe pliki logów. 

Na początek możemy „przeczytać” je do struktury, zwanej RDD: 

rdd_original = sc.textFile('/databricks-datasets/sample_logs/')
rdd_original.take(10)

Poprzednia komenda wykonuje odczyt wszystkich plików z podanej lokalizacji do struktury RDD. Z tak przygotowanym zestawem, możesz przekształcać dane (map/reduce/filter) w RDD, a także zbudować DataFrame na podstawie RDD. Niektóre elementy przekształcania danych zostały celowo pominięte w tych fragmentach kodu tylko po to, żeby uprościć ten artykuł. Cały “notebook” jest dostępny tutaj. Żeby stworzyć DataFrame, potrzebujemy jedynie przygotować schemę. 

from pyspark.sql.types import StructField, StructType, IntegerType, StringType, TimestampType

schema = StructType([
  # your data types definitions go here
])

df = rdd_mapped.toDF(schema)

Kiedy mamy już obiekt klasy DataFrame, droga jest otwarta na wszelkie agregacje i wyciągnie wniosków z naszych danych. I nie tylko użycie specyficznego API od Sparka wchodzi tu w grę. Można użyć zarówno jego, jak i SQL, jak w podanych tu przykładach. Przykładową informacją, którą możesz wyciągnąć, jest to, jak często dany użytkownik wywoływał akcje w Twoim systemie, które kończyły się błędem. 

df_identified = df[df['user'] != '-']
df_grouped = df_identified.groupby(['user','status']).count()

Dla Ciebie, mój drogi programisto pandas, te operacje powinny wyglądać znajomo, czyż nie? Oczywiście, jak już wspomniałem, możesz wyrazić to samo, korzystając ze składni SQL. 

df.createOrReplaceTempView('log')

select user, status, count(*)
from log
where user <> "-"
group by user, status

Z Databricks nie tylko dostajesz Sparka w chmurze, ale także bardzo prosto jest stworzyć proste wizualizacje. Szczególnie zachęcam Cię do używania ich dla Twoich celów badawczych. Nie bardzo nadają się do prezentacji danych dla Twojego klienta. 

Podsumowanie 

Analiza danych, która została tu wykonana, będzie jakiś czas publicznie dostępna pod tym linkiem. Możesz także ściągnąć notebooka stąd. Kończąc powiem, że ze Sparkiem i jego API DataFrame, możesz zrobić naprawdę dużo. Co najważniejsze, pełną krasę jego możliwości zobaczysz dopiero wtedy, kiedy będziesz zmagał się z naprawdę dużą ilością danych.