Integracja Git[Gitea] z Apache Airflow: Publikowanie i Zarządzanie DAG-ami

Apache Airflow przechowuje DAGi w katalogu 'dags'. Dokładną scieżkę do tego katalogu znajdziesz w pliku konfiguracyjnym Airflow. 

[core]
dags_folder = /opt/airflow/dags

Może się zdarzyć, że po instalacji ten katalog nie zostanie automatycznie utworzony - musisz go stworzyć. W najprostszym scenariuszu, publikacja nowego workflow (Dag) polega na umieszczeniu pliku w tym katalogu. Niestety ale Apache Airflow nie umożliwia publikacji z poziomu webowego GUI. W środowisku produkcyjnym takie zarządznie plikami DAGów byłoby jednak uciążliwe, dlatego lepszym pomysłem jest wykorzystanie systemu Git. Git jest jednym z przykładów oprogramowania z klasy „rozproszonych systemów kontroli wersji” (Distributed Version Control Systems, DVCS). 

Jeśli chcesz korzystać z systemu kontroli wersji, nie musisz ograniczać się do GitHub. Gitea, GitHub czy GitLab to serwery i GUI dla Git-a — korzystają z tych samych protokołów (SSH/HTTPS) i wywołują natywne komendy Git w tle, oferując interfejs webowy, zarządzanie użytkownikami, pull requesty i CI/CD, ale same nie zastępują Git-a. Dla naszych testów użyjemy lokalnego servera Git; Gitea.

Przygotowanie repozytorium w Gitea

W poniższych przykładach używać będziemy lokalnego serwera Gita zainstalowanego w środowisku WSL. Wcześniej jednak musimy zainstalować pakiet Git w środowisku WSL:

sudo apt update
sudo apt install git -y

Następnie tworzymy katalog (i podkatalogi) w którym będzie zainstalowana Gitea i nadajemy uprawnienia:

sudo mkdir -p /opt/gitea/{custom,data,log}
sudo chown -R $USER:$USER /opt/gitea
chmod -R 750 /opt/gitea

Teraz pobieramy binarkę ze strony Gitea, nadajemy plikowi prawo wykonywania i przenosimy go do docelowego katalogu:

wget -O gitea https://dl.gitea.io/gitea/1.24.6/gitea-1.24.6-linux-amd64
chmod +x gitea
sudo mv gitea /opt/gitea/

Tworzymy skrypt startujący Gitea; plik opt/gitea/start_gitea.sh:

#!/bin/bash
# -----------------------------
# Start Gitea in WSL
# -----------------------------

# Ścieżki
GITEA_BIN=/opt/gitea/gitea
GITEA_CUSTOM=/opt/gitea/custom
GITEA_WORK_DIR=/opt/gitea/data
LOG_DIR=/opt/gitea/logs
mkdir -p $LOG_DIR

# Funkcja uruchamia proces w tle jeśli nie działa
run_if_not_running() {
    local name=$1
    local cmd=$2
    if pgrep -f "$cmd" > /dev/null; then
        echo "$name is already running."
    else
        echo "Starting $name..."
        nohup $cmd > $LOG_DIR/${name}_$(date +%Y%m%d_%H%M%S).log 2>&1 &
    fi
}

# Eksport zmiennych środowiskowych
export GITEA_CUSTOM
export GITEA_WORK_DIR

# Start Gitea webserver
run_if_not_running "gitea-web" "$GITEA_BIN web --port 3000"
echo "Gitea startup script finished. Web UI: http://localhost:3000"

Za każdym razem kiedy będzie startować WSL, będzie uruchamiana Gitea - dodajemy więc komendę wykonującą ten plik do pliku bashrc:

nano ~/.bashrc
/opt/gitea/start_gite.sh

Więcej…

Czym są pliki w formacie Apache Parquet

Apache Parquet to format plików danych typu open source, zorientowany na kolumny, zaprojektowany z myślą o efektywnym przechowywaniu i pobieraniu danych. Oferuje wydajne schematy kompresji i kodowania, umożliwiające obsługę złożonych danych w dużych ilościach i jest obsługiwany przez wiele języków programowania oraz narzędzi analitycznych. Pełny opis formatu parquet oraz dokumentacja znajduje sie na stronie https://parquet.apache.org/

Czym się różni format CSV of formatu Parquet?

Główna różnica między plikami Parquet a CSV polega na sposobie przechowywania danych i efektywności: CSV to tekstowy format wierszowy, łatwy do odczytu przez człowieka, ale wolny przy dużych plikach, bez typów danych i zwykle większy.

Parquet to binarny format kolumnowy, wspierający kompresję i typy danych, umożliwiający szybki odczyt tylko potrzebnych kolumn i bardziej wydajny przy analizie dużych zbiorów danych.

CSV

Plik zwykle zaczyna się od nagłówka (pierwszego wiersza z nazwami kolumn), a kolejne wiersze zawierają dane, gdzie kolumny są oddzielone określonym separatorem, najczęściej przecinkiem, czasem średnikiem lub tabulatorem. Nagłówek nie jest obowiązkowy - dodawany jest tylko ze względu na łatwiejszy odczyt pliku. Oto przykład pliku tekstowego, CSV:

Parquet

Plik Parquet ma strukturę kolumnową i binarną: zamiast wierszy z separatorami, dane każdej kolumny są przechowywane razem w tzw. column chunks, a wiele takich bloków tworzy row groups. Plik zawiera także metadane opisujące schemat danych i typy kolumn, co umożliwia szybki odczyt wybranych kolumn. Nie ma tradycyjnego nagłówka ani separatorów – kolejność elementów w kolumnach odpowiada wierszom, dzięki czemu system wie, które wartości z różnych kolumn tworzą razem wiersz. Z naszego CSV powyżej powstały trzy kolumny.

Ale to nie wszystko. Dane są dzielone na tak zwane 'row groups' - bloki wierszy (np. po 50 tysięcy wierszy). Każdy taki zestaw wierszy zawiera 'chunki' (bloki kolumn) w których przechowywane są dane kolumn. Dane w plikach Parquet dzieli się na row groups, żeby szybciej wczytywać tylko potrzebne fragmenty, lepiej kompresować dane, przetwarzać je równolegle i efektywnie zarządzać pamięcią. Gdybyśmy mieli dwie row groups dla informacji z naszego CSV, jedna przechowywałaby pierwsze dwa wiersze a druga dwa następne:

Więcej…

Wprowadzenie do Apache Spark: uruchamianie zadań

Apache Spark to oprogramowanie, które pozwala sprawnie przetwarzać i analizować duże zbiory danych (naprawdę duże). Jest szybkie (bardzo szybkie!), skalowalne i wspiera różne języki programowania, dzięki czemu można go używać zarówno do prostych analiz, jak i bardziej zaawansowanych zadań. Spark ułatwia automatyzację pracy z danymi i pozwala szybciej uzyskać potrzebne wyniki.

Apache Spark może działać nie tylko na wydajnych serwerach - możesz je mieć nawet na swoim własnym laptopie i ten laptop może posłużyć Ci do całkiem zaawansowanych analiz. Dlatego warto zapoznać się z możliwościami tego oprogramowania.

Dwa sposoby uruchamianie zadań

Pomijamy tutaj uruchamianie skryptów pythona napisanych w pyspark (jeśli zaistalujesz Spark jako moduł pyspark do pythona). Moduł PySpark zainstalowany przez pip działa w trybie lokalnym i uruchamia Spark tylko na pojedynczym komputerze, pod pełną kontrolą Pythona, bez możliwości użycia języka Scala ani pełnego rozproszonego przetwarzania. Dlatego sprawdza się głównie do testów, nauki i eksperymentów, ale nie do przetwarzania dużych zbiorów danych w produkcji.

spark-shell (pyspark)

Spark-shell stosuje się, gdy chcemy interaktywnie testować i eksplorować dane w Sparku przy użyciu Scali, bez konieczności pisania pełnej aplikacji. Jest idealny do szybkiego prototypowania, nauki Sparka i eksperymentów z operacjami na RDD (Resilient Distributed Dataset) czy DataFrame.

Nasz plik z danymi do testów ma strukturę (pobierz plik):

customer_number;sales_amount;sales_date
9;4089.54;2024-12-13
18;6579.65;2025-04-19
20;6005.84;2024-08-02
7;2026.39;2024-12-28
20;6870.04;2024-01-12
17;4454.97;2025-04-10
5;1116.2;2024-07-28
15;1463.37;2025-04-13
3;1229.04;2024-08-20
1;2397.96;2024-06-08
5;4005.88;2025-02-04
8;8435.51;2024-03-15
16;6749.28;2023-10-22
3;9665.37;2024-10-26
4;5635.03;2025-09-14
1;8108.14;2024-09-29
5;1522.14;2025-06-15
11;9610.69;2025-01-24
3;6675.06;2025-06-11
2;4545.24;2025-05-30

Plik zapisz do dowolnego katalogu na swoim PC. Środowisko WSL umożliwia odczyt plików  z katalogów Windows jakby to był ten sam system operacyjny. Jeśli katalogiem w którym masz ten plik jest dysk G:, pełna ścieżka 'G:\ApacheSpark\Data', dla WSL będzie to scieżka '/mnt/g/ApacheSpark/Data/'. Sprawdź to wpisując w shell'u WSL następujące polecenie:

ls /mnt/g/ApacheSpark/Data/

W ten sposób otrzymasz listę plików które znajdują się w katalogu. Jeśli masz już uruchomione okno WSL (poprzez Powershell), uruchom shell Spark wpisując spark-shell. Kiedy uruchamiasz spark-shell, Spark automatycznie włącza REPL (Read-Eval-Print Loop) dla Scali. Spark-shell jest REPL-em tylko języka Scala. Jeśli masz kod Python'a, uruchom REPL który obsługuje skrypty pyspark wpisujac pyspark w shellu WSL.

Zróbmy pierwsze ćwieczenie - odczytajmy pierwszych 10 wierszy z powyższego pliku wklejając kod napisany w Scala bezpośrednio do okna spark-shell:

// Utworzenie SparkSession
val spark = org.apache.spark.sql.SparkSession.builder()
  .appName("Odczyt CSV")
  .getOrCreate()

// Ścieżka do pliku w WSL
val plik_csv = "/mnt/g/ApacheSpark/Data/sales_data.csv"

// Wczytanie CSV z separatorem ';' i nagłówkiem
val df = spark.read
  .option("header", "true")
  .option("sep", ";")
  .csv(plik_csv)

// Wyświetlenie pierwszych 10 wierszy
df.show(10)

// Zakończenie sesji Spark
spark.stop()

Więcej…