Gdybym miał wymienić dwa najpopularniejsze “buzzwordy” związane z chmurą publiczną to byłoby to bigdata i serverless. Ten pierwszy jest już obecny wśród nas od wielu lat i zyskał na popularności wtedy, gdy przetwarzanie dużej ilości danych stało się łatwiejsze poprzez większą dostępność do mocy obliczeniowej. Chociażby właśnie z wykorzystaniem chmury publicznej.

Z kolei Serverless jest określeniem dużo młodszym, natomiast z wielkim impetem wpadł do świata IT, szczególnie do środowiska deweloperów. Warunek: W 99% przypadków potrzebujemy chmury, żeby tworzyć rozwiązania „bez serwerów”.

Dziś chciałbym pokazać jak można oba te określenia połączyć, czyli stworzymy system przetwarzający duże ilości danych uruchamiany w całości w modelu Serverless.

Zacznijmy od zdefiniowania problemu

To co opisuję poniżej nie jest wymyślonym, hipotetycznym problemem, lecz rozwiązaniem przygotowanym na potrzeby jednego z naszych klientów. W projekcie chodziło o to, żeby przetwarzać duże pliki .csv wrzucane do kontenera w usłudze Azure Blob Storage. Pliki były umieszczane z różną częstotliwością, ale mniej więcej jeden raz dziennie. Sama analityka odbywała się przy pomocy narzędzia Apache Spark.

Oczywiście klientowi zależało, żeby system był możliwie tani, dlatego szybko odrzuciliśmy pomysł na użycie uruchomionych non-stop wirtualnych maszyn z zainstalowanym Apache Spark. Z podobnego powodu nie zdecydowaliśmy się ma rozwiązanie Azure HDInsight. Na szczęście z pomocą przyszedł Azure Databricks.

Czym jest Azure Databricks?

Jest to usługa z kategorii bigdata, służąca do analizy dużych zbiorów danych. Możemy w niej zarówno przetwarzać dane statyczne, jak i strumieniowe, mamy również narzędzia do implementacji modeli uczenia maszynowego. Całość oparta jest o popularny framework Apache Spark, więc jeśli go znasz to zdecydowanie zachęcam, żebyś zainteresował się Azure Databricks.

Ponadto, usługa bardzo dobrze integruje się z innymi serwisami Azure. Dzięki temu bez problemu możemy chociażby czytać dane z Azure Blob Storage czy też zarządzać użytkownikami przy pomocy Azure Active Directory.

Usługa jest łatwo dostępna dla szerokiego grona deweloperów i analityków. Dlaczego? Po pierwsze dlatego, że dzięki odpowiednim interfejsom kod może być pisany w jednym z czterech języków do wyboru: Scala, Python, R oraz SQL. Po drugie Azure Databricks pozwala na łatwą współpracę dzięki wykorzystaniu współdzielonych, interaktywnych notebooków.

Architektura rozwiązania

Wracając do naszego klienta, zaproponowaliśmy następująca architekturę jego systemu. To co widzisz poniżej jest lekko uproszczonym diagramem przygotowanym na potrzeby tego artykułu:

Mamy więc konto Azure Storage, na którym będą lądować pliki do analizy. Każdy taki plik będzie wywoływał Azure Function, która z kolei będzie uruchamiać joba Sparkowego w Azure Databricks. Job ten pobierze sobie plik bezpośrednio z Blob Storage i przetworzy go według własnej logiki biznesowej.

Jakie są zalety takiego rozwiązania:

• Nie utrzymujemy stale żadnych serwerów – 100% serverless
• Płacimy tylko za zużycie, jeśli rozwiązanie jest nieużywane to nie płacimy nic
• Koszty rosną proporcjonalnie do stopnia wykorzystania
• System jest typu „event-driven”, więc sam się skaluje

Zapraszam Cię zatem na krótki przewodnik o tym jak takie rozwiązanie szybko i łatwo samemu zbudować!

Tworzymy Azure Databricks

Zacznijmy od stworzenia przestrzeni roboczej Azure Databricks. Pozwoli ona na uruchamianie klastrów obliczeniowych oraz Jobów Apache Spark. Logujemy się do portalu Azure, tworzymy nowy zasób o nazwie Azure Databricks

Zatwierdzamy nasz wybór i czekamy kilka minut. 

Po utworzeniu usługi otwieramy ją. Jak możesz zauważyć praca z Databricks odbywa się w osobnym panelu, z poziomu którego będziemy zarządzać wszystkimi elementami usługi. Dobra integracja z Microsoft Azure sprawie, że tego portalu logujemy się naszym kontem Azure Active Directory. 

Tworzymy joba Apache Spark

Żeby utowrzyć joba potrzebujemy mieć kod napisany w Apache Spark. Jeśli taki mamy, możemy utworzyć joba przykładowo na podstawie pliku .jar. My jednak nie posiadamy żadnego kodu, dlatego utwórzmy nowego interaktywnego notebooka. Co to takiego? Jest to takie „IDE”, w którym możemy pisać kod i na bieżąco widzieć efekty jego wykonania. Przykładowo wczytać dane z pliku .csv, a następnie je podejrzeć lub zwizualizować. Co ważne, na jednym takim notebooku może współpracować kilka osób.

Klikamy więc przycisk „New Notebook” i wypełniamy prosty formularz:

Wybieramy język programowania, w którym będziemy pisać kod, jak już wspominałem do wyboru jest Python, Scala, R oraz SQL. Ja preferuję Pythona. Możemy również wybrać klaster, na którym będzie uruchamiany nasz kod, jeśli klastra jeszcze nie mamy to możemy go stworzyć później.

Kolejne zadanie to zaimplementowanie całej logiki przetwarzania danych dla naszego joba

Oczywiście ja nie mogę tutaj zamieścić kodu, który przygotowaliśmy dla naszego klienta, ale pokażę dwa jego istotne fragmenty

Odczyt plików z Azure Blob Storage

Pierwsza pytanie: Jak w Azure Databricks odczytać pliki z Blob Storage? Robi się po poprzez zamontowanie kontenera w systemie DBFS (Databricks File System – system plików tworzony na potrzeby klastra Azure Databricks). Dzięki temu mamy dostęp z poziomu Databricks bezpośrednio do konta Blob Storage Poniżej przykład kodu, który za to odpowiada:

dbutils.fs.mount(
      source="wasbs://@"),
      mount_point="/mnt/blobfiles",
      extra_configs = {"fs.azure.account.key..blob.core.windows.net":""}
    )

Przekazywanie parametrów do joba

Wiemy już jak odczytać plik z Blob Storage, ale pytanie: skąd job ma wiedzieć który plik odczytać? Chcemy przetworzyć ten plik, którego umieszczenie w kontenerze sprawiło, że uruchomiła się nasza funkcja i nasz job. Jednak tylko funkcja wie dokładnie jaki to był plik. Na szczęście w Databricks Joby mogą przyjmować parametry przy uruchamianiu. Wartości tych parametrów możemy więc łatwo odczytać w kodzie w ten sposób:

fileName = dbutils.widgets.get("fileName")

Powyższa linijka odczyta wartość parametru o nazwie „fileName”. Dzięki temu dokładnie wiemy, który plik powinniśmy odczytać i przetworzyć.

Mając oba powyższe fragmenty kodu możemy wczytać plik, który znajduje się na Blob Storage jako DataFrame:

df = spark.read.format("csv")
.option("inferSchema", "true")
.option("header","false")
.load("dbfs:/mnt/blobfiles/" + fileName)

Tworzymy instancję Joba

Kolejny krok to zdefiniowanie Joba na poziomie Azure Databricks. W tym celu w panelu po lewej stronie klikamy przycisk „Jobs”, a następnie „Create Job”. W nowym oknie musimy wypełnić kilka parametrów:

• Task – tu de facto wskazujemy kod Joba. W naszym przypadku wybieramy nowo utworzonego notebooka. W tym miejscu możemy podać również parametry. Nas to jednak interesuje, gdyż wartość parametru fileName będzie zależna od nazw umieszczanych na Blob Storage plików
• Cluster – na potrzeby wykonanie Joba zostanie użyty klaster Sparkowy. Mamy tak naprawdę dwie opcje: albo tworzyć nowy klaster za każdym razem, kiedy Job jest wywoływany, albo użyć już istniejący. Generalnie zaleca się opcję pierwszą, gdyż taki klaster automatycznie wyłączy się po zakończeniu procesowania. Będzie też on inaczej rozliczany (korzystniej)
• Schedule – naszego Joba możemy również uruchamiać cyklicznie. W naszym przypadku nie włączamy tej możliwości

Poniżej przykład konfiguracji:

Mamy już prawie wszytsko dopięte po stronie Azure Databricks. 

Prawie, bo pozostaje pytanie: W jaki sposób wywołamy joba z poziomu Azure Functions?

Da się to zrobić poprzez udostępnione API Restowe, które pozwala wykonywać różnorakie operacje w Databricks, m.in. właśnie uruchomienie Joba. Żeby wywołać API potrzebujemy:

• Endpointa
• Tokenu uwierzytelniającego
• Id utworzonego Joba

Endpoint jest łatwo dostępny z poziomu portalu Azure:

Z kolei, żeby utworzyć token, otwieramy panel Databricks, w prawym górnym rogu klikamy w ikonę ludzika i wybieramy opcję „User Settings”. Następnie przy pomocy guzika „Generate New Token” generujemy nowy token zapisujemy go w bezpiecznym miejscu.

Id utworzonego Joba łatwo odnajdziemy w jego parametrach.

Jeśli chodzi o konfigurację, po to stronie Azure Databricks to tak naprawdę wszystko.

Piszemy Azure Function

Mamy już przygotowane środowisko Databricks dlatego teraz pora na kolejny klocek naszej układanki, czyli Azure Functions. Kod funkcji będzie dość prosty, musimy w niej:

1. Przechwycić nazwę pliku wrzuconego na Blob Storage
2. Wywołać naszego Joba w Azure Databricks

Tworzymy zatem nową Function App, a następnie dodajemy funkcję triggerowaną obiektem w Blob Storage:

Kod funkcji nie będzie ani długi, ani skomplikowany, poniżej najważniejsze jego elementy:

1. W pierwszym kroku pobieramy wszystkie niezbędne parametry ze zmiennych środowiskowych:

string token = Environment.GetEnvironmentVariable("DatabricksToken");
string url = Environment.GetEnvironmentVariable("DatabricksURL") + "/api/2.0/jobs/run-now";
string jobIdString = Environment.GetEnvironmentVariable("DatabricksJobID");

2. Następnie definiujemy obiekt, który wyślemy jako zawartość requestu do API:

class JobBody
{
    public int job_id {get; set;}
    public NotebookParams notebook_params {get; set;}
}
 
class NotebookParams
{
    public string fileName {get; set;}
}

3. Tworzymy instancję klasy JobBody i wypełniamy ją parametrami:

var body = new JobBody();
body.job_id = Int32.Parse(jobIdString);
var notebookParams = new NotebookParams();
notebookParams.fileName = fileName;
body.notebook_params = notebookParams;

Zmienna fileName jest pobierana z argumentu funkcji „Run” Azure Functions. Jest to nazwa pliku umieszczonego na Blob Storage

4. Tworzymy klienta http i wysyłamy zapytanie do API Databricks:

var client = new HttpClient();
client.DefaultRequestHeaders.Add("Authorization","Bearer " + token);
var response = await client.PostAsync(url, body.AsJson());
var contents = await response.Content.ReadAsStringAsync();
public static StringContent AsJson(this object o)
 => new StringContent(JsonConvert.SerializeObject(o), Encoding.UTF8, "application/json");

Na koniec nie pozostaje nam nic innego jak przetestować nasze rozwiązanie. Wystarczy umieścić plik na Blob Storage i patrzeć jak cała reszta dzieje się samoczynnie J

Podsumowanie

Przy pomocy bardzo prostej architektury stworzyliśmy system mogący przetwarzać duże dane wydajnie i skalowalnie. Jest to o tyle atrakcyjne, że wprowadzenie takiego rozwiązania nie wymaga nakładu dużych środków finansowych na maszyny (czy to w chmurze czy on-premise), na których instalowalibyśmy Apache Spark. Jeśli więc potrzebujemy systemu bigdata, to łatwo i tanio w ten sposób zacząć. Pamiętajmy jednak zawsze o monitorowaniu wydajności takiej architektury jak i kosztów, które będą rosnąć wraz z użyciem.

Jeśli chciałbyś dowiedzieć się więcej o tym jak budować tego typu architektury, służące do przetwarzania dużej ilości danych, to zapraszamy Cię bardzo serdecznie na nasze praktyczne szkolenie on-line, dzięki któremu będziesz mógł zostać inżynierem danych w chmurze Microsoft Azure: https://szkolachmury.pl/kursy/azure-data-engineer/

Linki:

Azure Databricks: https://azure.microsoft.com/pl-pl/services/databricks/

Apache Spark: https://spark.apache.org/

Dokumentacja Azure Databricks: https://docs.azuredatabricks.net/index.html

Azure Functions: https://azure.microsoft.com/pl-pl/services/functions/