Jak używać funkcji okienkowej w Spark?

Rafał Gogłoza

Problem:

SQL jest prosty. Funkcje okienkowe są proste, intuicyjne i przychodzi bez komplikacji. Natomiast w Spark? Jak wytłumaczyć koledze w zespole w jaki sposób tworzyć funkcję okienkowe w Spark? Od czego zacząć? Jakie są problemy i wyzwania w tworzeniu funkcji okienkowych? Jakie są różnic w porównaniu z SQL? Co jest łatwiejsze w utrzymaniu?

Rozwiązanie:

Na szybko można odpowiedzieć w ten sposób, że funkcja okienkowa tworzona jest w oderwaniu od data setu ale aplikowania na jego podzbiorze. Na przykład definicja wygląda w ten sposób:

user_window = Window.partitionBy(„user_id”).orderBy(„event_time”)

Jak to rozumieć? Dla każdego użytkownika okienko będzie posortowane po even_time i najstarsze dane będą pojawiały się na jako pierwsze a najświeższe dane wylądują w tym zbiorze danych jako ostatnie.

Potem wystarczy już tylko użyć funkcji okienkowej.

Ale zacznijmy po kolei i na przykładzie.

Read More

Katastrofa: cluster działał ponad 24h

Rafał Gogłoza

Problem:

Ładowanie rozpoczęło się w weekend a w poniedziałek rano stwierdziliśmy, że cluster wciąż działa. Ładowanie, które rozpoczęło się w niedzielę rano, trwało ponad 24h i w poniedziałek rano ładowanie wciąż było w trakcie. W sumie ładowanie powinno trwać 3h, trwało ponad 21h dłużej i nie zakończyło się sukcesem. Pociągnęło to za sobą duże koszty i nasunęło dwa pytania: co się stało, dlaczego cluster działa tak długo? Co zrobić, żeby to się więcej nie powtórzyło.

Rozwiązanie (a raczej przyczyna)

Spojrzenie w szczegóły ładowania pokazały, że był jeden notebook, gdzie ładowanie trwało ponad 21h i wciąż trwało. Jeżeli właśnie pomyślałeś, że to duplikaty to masz rację. Dane zostały zwielokrotnione. Dla niektórych rekordów z tabeli źródłowej było 32 768 rekordów. Nie spowodowało to wywrócenia procesu i błędu wskazującego na nie wystarczającą pamięć (out of memory exception). Ładowanie trwało ale trzeba było je przerwać, nie chcieliśmy go kontynułować, gdyż nie zawierało poprawnych danych.

Dobrze wiemy już co się stało, w takim razie jak przeciwdziałać takim sytuacjom w przyszłości?

Read More

Ładowanie danych z Databricks do Azure Synapse Analytics

Rafał Gogłoza

Problem:

Zadanie zostało zdefiniowane przez managera w ten sposób:
– Dane z Databricks mają zostać przesłane na Azure Synapse
– Security utworzyło i otworzyło odpowiednie private endpointy.
– Dostałeś też namiary na service principala, którego wykorzystasz do zapisywania danych z Databricks na Azure Synapse.

Będziesz używał sparka, żeby od razu stworzył tabelę i dane. Będzie to szczególnie pomocne gdyż danych nie jest zbyt dużo. Nie powinno być żadnych problemów wydajnościowych.
Niestety pojawiają się problemy zupełnie innej natury. Przy próbie wstawienia danych dostajesz błąd:


„com.microsoft.sqlserver.jdbc.SQLServerException: The statement failed. Column 'drone_spec_key’ has a data type that cannot participate in a columnstore index.”

Rozwiązanie:

Spark wysyłał do Synapsa create i insert statement w tym samym czasie. Błąd wynikał z tego, że Synapse przy próbie stworzenia tabeli jednocześnie próbuje stworzyć custered index. Niestety ograniczenie, które posiada to brak możliwości stworzenia indeksu na kolumnach, gdzie typ danych zdefiniowany jest jako: VARCHAR(max), NVARCAHR(MAX) a to się dzieje, gdy spark próbuje stworzyć tabelę.
Jako rozwiązanie zastosowano:
1. Najpierw została tworzona tabela po stronie Azure Synapse Analytics
2. Dopiero później zostały wstawione do niej dane

Szczegóły kodu oraz alternatywne, dające więcej możliwości, rozwiązanie poniżej.

Read More

Qualify w Databricks

Rafał Gogłoza

Problem:

Działasz na tabeli opisującej procesy produkcji. Jeden proces może występować więcej niż jeden raz w tabeli. Nas interesuje tylko ostatnia data zakończenia procesu. Użyjemy funkcji okienkowej, row_number, żeby oznaczyć proces, który zakończył się jako ostatni. W jaki sposób w tym samy zapytaniu wybrać ten wiersz, bez używania dodatkowych podzapytań lub CTE?

Rozwiązanie:

Użyjemy Qualify! Qualify w Databricks filtruje wyniki zapytania funkcji okienkowej. Możesz myśleć o nim jak warunku zakładanym na wyniku funkcji okienkowej. Upraszcza to znakomicie składnie.

Jak go zastosować:

select process_id, process_name, process_start_date, process_end_date
, row_number() over(partition by process_id order by process_end_date desc) as rn
from d_process
qualify rn = 1

Read More

Python i Synapse: połączenie przy użyciu service principala

Rafał Gogłoza

Problem:

Masz bazę danych Synapse dostępną w usłudze Azure. Posiadasz uprawnienia, żeby połączyć się do tej bazy przy użyciu service principala. To połączenie musisz wykonać w Pythonie. W jaki sposób połączyć się posiadając dane service principala do usługi Synapse?

Rozwiązanie:

Użyjemy .env do przechowywania zmienny środowiskowych. Użyjemy azure.identity żeby połączyć się z usługą i wygenerować token dostępowy. Następnie posiadając token utworzymy połączenie pyodbc.

Zawiłe? Teraz trochę bardziej szczegółowo.

Read More

Databricks: porównanie CTAS, deep i shallow clone

Rafał Gogłoza

Problem:

Masz już wyczyszczone, połączone, załadowane i gotowe do konsumpcji przez warstwę raportową dane. Teraz Power BI będzie pobierało dane do dataflow. Następne ładowanie zaczyna się za chwilę i zmieni, kształt danych. Chcesz jednak załadowane dane wysłać do raportów. Żeby zapewnić spójny obraz danych potrzebujesz zrobić snapshot (migawkę) danych. Potrzebujesz zduplikować istniejące dane. Masz do wyboru: create table as select (CTAS), deep (głęboki) oraz shallow (płytki) klon danych.

Rozwiązanie:

W moim przypadku najlepszym rozwiązaniem będzie użycie głębokiego klonowania (deep clone). Płytkie klonowanie ma niestety ograniczenia, które wykluczają jego zastosowanie. CTAS jest zbyt wolny, to było dopiero odkrycie! Tabela, na której testowałem nie jest duża ma 0,5 GB ale to wystarczający wolumen żeby wyciągnąć wnioski.

Read More

Databricks: Jak znaleźć wolno działający notebook?

Rafał Gogłoza

Problem:

Czas procesowania danych w Databricks zwiększył się dwukrotnie. Poprzednio wynosił 3 godziny teraz wynosi 6. Zanim podejmiesz proces naprawy trzeba sprawdzić który notebook spowodował aż tak duży spadek wydajności. Czy spadek wydajności rozlał się równomiernie pomiędzy wszystkie notebooki? Czy dotyczy tylko części? Możemy wykluczyć, że ktoś inny pracował na platformie w tym samym czasie i wpływał na wydajność. Cluster jest dedykowany do przetwarzań batchowych i nikt inny nie ma do niego dostępu.

Rozwiązanie:

Użyjemy Databricks rest API, żeby przeszukać wszystkie joby i znaleźć ten, który trwał najdłużej w porównaniu z poprzednim ładowaniem. Gdy znajdziemy, który to job, wtedy przeszukamy wszystkie taski i sprawdzimy czy są jacyś pojedynczy kandydaci, których wydajność zdecydowanie spadła i znajdziemy wolno działający notebook.

Read More

Azure storage file datalake do pobierania plików?

Rafał Gogłoza

Problem:

Masz dostępne Databricks, Pythona i Azure Storage Account. Potrzebujesz pobrać raport Power BI umieszczony na Azure Storage account przy pomocy Pythona.
Potem ten plik należy opublikować w serwisie Power BI.
Nie możesz tego zrobić przy użyciu Sparka, albo Pandas. To się nie uda i jednocześnie, to nie jest to zadanie.
Możesz instalować bilbioteki na clustrze. Najlepiej, żeby ich autorem był Microsoft.

Rozwiązanie:

Microsoft udostępnia bibliotekę: azure.storage.filedatalake przy pomocy której można przeczytać plik z landing zone w formacie binarnym a potem opublikować go w portalu Power BI.

Wystarczy z kontenera przeczytać plik. Ta zawartość zostanie wczytana w formacie binarnym:
file_content = file_container.download_file()

A potem opublikować go w Power BI portalu:
publish_powerbi_report(PBI_WORKSPACE_ID, PBI_REPORT_NAME, file_content)

Read More

Databricks: Jak pobrać pliki binarne z Azure Storage Account używając Pythona?

Rafał Gogłoza

Problem:

Masz dostępne Databricks, Pythona i Azure Storage Account. Potrzebujesz pobrać plik z Azure Storage account przy pomocy Pythona w formacie binarnym. Jeżeli chcesz zrobić to przy użyciu Sparka, albo Pandas to nie jest to zadanie.
Dane masz pobrać z pliku binarnego.
Trzeba użyć modułów Pythona do wczytywania plików.
W dokumentacji piszą, że taka operacja jest „not supported”. (Stan na 15.10.2024)
Nie chcesz też robić „mount” zdalnego systemu plików. Taka operacja jest nie polecana przez Databricks.

Rozwiązanie:

W Databricks z Pythona NIE można czytać plików ze zdalnego systemu plików. Można za to czytać pliki z lokalnego file systemu. Obejście problemu przedstawionego powyżej to:
1. Przy użyciu dbutils.fs albo %fs skopiować pliki ze zdalnego filesystemu do lokalnego.
2. Przeczytać pliki z lokalnego systemu plików przy użyciu Pythona.
Skasować skopiowany plik z lokalnego filesystemu.

Read More

Databricks: Jak ustawić własną wartość dla kolumny identity?

Rafał Gogłoza

Tło wydarzeń:

Masz tabelę stworzoną w Databricks. Klucz główny ustawiony jest jako kolumna identity. Dzięki temu masz obsłużone wstawianie unikalnych wartości. Nie musisz też tego robić samodzielnie, co tworzy miejsce na pominięcie czegoś. Jednak pojawia się wyzwanie. Twoja „sekwencja” zaczynała wstawianie danych od jedności a teraz chciałbyś wstawić singletony (-1, -2).

Problem:
Chciałbyś wstawić singletony do tabeli, gdzie jedna z kolumn jest typu Identity. W jaki sposób wstawić do niej oczekiwane wartości?

Rozwiązanie:
Zdefiniuj kolumnę klucza głównego jako:

generated by default as identity

Ale uważaj, jest jeden przypadek, gdy tak zdefiniowana kolumna może sprawić problemy.

Read More