Odczyt i zapis danych podczas uruchamiania zadań Apache Spark w środowisku Kubernetes

Jeśli chcemy analizować dane w Apache Spark w środowisku Kubernetes musimy zadbać o odczyt danych i zapis rezultatu po wykonaniu zadania. Tego problemu nie ma kiedy używamy Apache Spark w formie Standalone i mamy tylko jeden worker; możemy odczytywać dane i je zapisywać na dysk bez dbałości o widoczność tego zaspobu przez wiele executorów Sparka.

Kiedy Apache Spark działa w klastrze 'standalone' i uruchamianych jest wiele workerów, najczęstszy scenariusz to kopiowanie danych do analiz na każdą maszynę workera i zapisywanie rezultatów na każdej z maszyn a po wykonanej pracy ich 'merge' z workerów do finalnego rezultatu. Wbrew powszechnemu marketingowi, Apache Spark nie pracuje najcześciej z danymi RAW/BRONZE czy SILVER na setkach Terrabajtów czy na Petabajtach danych. Najczęściej takie dane są dzielone i workery otrzymują tylko część danych lub danych w oknie czasowym jest stosunkowo niewiele. Dopiero zadania na danych GOLD wykonywane są na dużych wolumenach danych.

Pody Apache Spark w Kubernetes są tworzone dynamicznie i przestają istnieć po wykonaniu zadania. Dystrybuowanie do nich danych i ich zapis byłby problematyczne. Teoretycznie raczej mamy dwa scenariusze:

Dane lokalne w podzie – każdy pod widzi tylko swój lokalny filesystem. Mówimy tylko o zapisie danych do lokalnego poda bo skoro pod tworzony jest dynamicznie, problematyczne jest przesłanie do niego danych.
Sieciowy PersistentVolume (PV, PVC, ReadWriteMany, np. NFS, Ceph, EFS) – wszystkie pody (driver + executory) montują ten sam katalog, dane są widoczne wszędzie; Spark może odczytywać i zapisywać równolegle. Dane przesyła się po prostu zapisując je w PV; można też wstępnie wrzucić pliki do PV przed startem joba.

Oba scenariusze są trudne w produkcyjnym wykorzystaniu. Wskazane jest by wszystkie workery widziały ten sam zasób który jednocześnie jest łatwy do zarządzania. Dlatego używamy trzeciego rozwiązania:

Obiektowy storage (S3, GCS, Azure Blob) – wszystkie pody mają dostęp przez protokoły S3/HTTP, Spark odczytuje i zapisuje dane w formacie np. Parquet, CSV; dane przesyła się do bucketu przed startem joba, np. aws s3 cp /local/path s3://bucket/input.

Amazon Simple Storage Service (Amazon S3) lokalnie

Najprostszym, nie najbardziej wydajnym, jest używanie usługi która wykorzystuje właśnie S3 (protokół s3, s3a). Dzięki https://www.min.io/ taki serwis możesz mieć lokalnie. Oprogramownie jest proste do instalacji. Oprogramowanie udostępnia tę sama funkcjonalność która daje chmura AWS - za darmo. Istnieje też wersja bardziej rozbudowana, komercyjna oraz, a jakże, chmura MinIO.

Firma MinIO zrobiła co prawda klasyczny open-core shift i usuneła wiele funkcji ostatnio ze swojej darmowej wersji oprogramowania. I tak by zarządzać uprawnieniami potrzebny jest client który pobierzesz tutaj. Jednak do jeśli potrzebujesz serwisu S3 do stosunkowo niewielkich volumenów danych, MinIO jest wszystkim czego potrzebujesz. Możliwa jest także budowa trybu rozproszonego (distributed mode) w MinIO Community.

MinIO (i Amazon S3) są przechowywane w buckets. Możliwe jest stworzenie 'katalogów' w każdym z bucket które jednak nie są prawdziwymi katalogami ale 'kluczami'. Klucz (key) w S3/MinIO to cały ciąg znaków, który jednoznacznie identyfikuje obiekt w buckecie.

Może wyglądać jak ścieżka z /, np. photos/2025/img001.jpg,
ale / to tylko znak w nazwie — nie tworzy prawdziwych katalogów,
cały ciąg jest jednym kluczem, nawet jeśli wygląda jak kilka „folderów”.

Czyli w --key "photos/2025/" kluczem jest cały ciąg "photos/2025/".

To co widzisz w konsoli, wygląda jak typowy system plików z folderami.:

mybucket/
└── photos/
    ├── 2024/
    │   └── winter/
    │       └── img001.jpg
    └── 2025/
        ├── summer/
        │   ├── img001.jpg
        │   └── img002.jpg
        └── winter/
            └── img003.jpg

W rzeczywistości bucket mybucket zawiera tylko płaską listę obiektów z nazwami (kluczami):

mybucket:
  photos/2024/winter/img001.jpg
  photos/2025/summer/img001.jpg
  photos/2025/summer/img002.jpg
  photos/2025/winter/img003.jpg

Odczyt i zapis do serwisu S3 (MinIO)

Poniżej przykład jak w poleceniu spark-submit wskazać serwis S3 (protokół S3a). Ponieważ poleceniem spark-submit nie możesz wysłać kodu programu, wskazujemy gdzie on jest w jednym z parametrów: Przykład dla wersji Apache Spark 4 (uruchomienie przykładowego zadania napisanego w Python):

spark-submit \
  --master k8s://https://10.0.0.120:6443 \
  --deploy-mode cluster \
  --name minio-sales-test \
  --conf spark.kubernetes.namespace=spark \
  --conf spark.kubernetes.authenticate.driver.serviceAccountName=spark \
  --conf spark.executor.instances=3 \
  --conf spark.driver.memory=4g \
  --conf spark.executor.memory=4g \
  --conf spark.driver.cores=1 \
  --conf spark.executor.cores=1 \
  --conf spark.kubernetes.container.image=apache/spark:4.1.0-preview3-python3 \
  --conf spark.driver.extraJavaOptions="-Divy.home=/tmp/.ivy2" \
  --conf spark.executor.extraJavaOptions="-Divy.home=/tmp/.ivy2" \
  --packages org.apache.hadoop:hadoop-aws:3.4.2,com.amazonaws:aws-java-sdk-bundle:1.12.792 \
  --conf spark.hadoop.fs.s3a.endpoint=http://10.0.0.120:9000 \
  --conf spark.hadoop.fs.s3a.access.key=admin \
  --conf spark.hadoop.fs.s3a.secret.key=xxx \
  --conf spark.hadoop.fs.s3a.path.style.access=true \
  --conf spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem \
  s3a://scripts/test.py

Zauważ, że mamy tutaj parametr '--packages org.apache.hadoop:hadoop-aws:3.4.2,com.amazonaws:aws-java-sdk-bundle:1.12.792' To polecenie mówi Sparkowi, żeby pobrał i dołączył z Maven dodatkowe biblioteki: hadoop-aws (łączenie z S3) i aws-java-sdk-bundle (SDK do komunikacji i uwierzytelnienia), dzięki czemu Spark może czytać i zapisywać dane w S3 (s3a://). Standardowy obraz Apache Spark nie zawiera wszystkich bibliotek Hadoop, zwłaszcza tych potrzebnych do obsługi S3.

W podzie poniższe parametry ustawiają Sparkowi i JVM miejsce, gdzie Ivy pobiera paczki Maven (--packages) — zamiast domyślnego ~/.ivy2 używa /tmp/.ivy2, który jest zapisywalny w kontenerze, dzięki czemu driver i executory mają dostęp do pobranych JAR-ów:

--conf spark.driver.extraJavaOptions="-Divy.home=/tmp/.ivy2" \
--conf spark.executor.extraJavaOptions="-Divy.home=/tmp/.ivy2" \

Polecenie spark-submit dla wersji Apache Spark 3.5.7:

spark-submit \
 --master k8s://https://10.0.0.120:6443 \
 --deploy-mode cluster \
 --name minio-sales-test \
 --conf spark.kubernetes.namespace=spark \
 --conf spark.kubernetes.authenticate.driver.serviceAccountName=spark \
 --conf spark.executor.instances=3 \
 --conf spark.driver.memory=4g \
 --conf spark.executor.memory=4g \
 --conf spark.driver.cores=1 \
 --conf spark.executor.cores=1 \
 --conf spark.kubernetes.container.image=apache/spark:3.5.7-scala2.12-java17-python3-r-ubuntu \
 --conf spark.driver.extraJavaOptions="-Divy.home=/tmp/.ivy2" \
 --conf spark.executor.extraJavaOptions="-Divy.home=/tmp/.ivy2" \
 --packages org.apache.hadoop:hadoop-aws:3.3.4,com.amazonaws:aws-java-sdk-bundle:1.12.262 \
 --conf spark.hadoop.fs.s3a.endpoint=http://10.0.0.120:9000 \
 --conf spark.hadoop.fs.s3a.access.key=admin \
 --conf spark.hadoop.fs.s3a.secret.key=xxx \
 --conf spark.hadoop.fs.s3a.path.style.access=true \
 --conf spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem \
  s3a://scripts/test.py

Szczegóły parametrów przesyłanych w spark-submit

--conf spark.driver.extraJavaOptions="-Divy.home=/tmp/.ivy2"
--conf spark.executor.extraJavaOptions="-Divy.home=/tmp/.ivy2"

Mówią Javie, żeby używała katalogu /tmp/.ivy2 jako lokalnej pamięci podręcznej Ivy (systemu zarządzania zależnościami Spark).
To eliminuje błędy typu /nonexistent/.ivy2/... i pozwala Sparkowi zainstalować pakiety (np. hadoop-aws) w kontenerach.

--packages org.apache.hadoop:hadoop-aws:3.3.4,com.amazonaws:aws-java-sdk-bundle:1.12.262

Mówi Sparkowi, żeby pobrał i dodał do classpath te dwa pakiety JAR:

hadoop-aws – adapter pozwalający Hadoopowi/Sparkowi komunikować się z S3.
aws-java-sdk-bundle – biblioteka AWS (obsługa autoryzacji, S3 API, itp.).

Bez tego Spark nie wiedziałby, jak obsłużyć s3a://.

--conf spark.hadoop.fs.s3a.endpoint=http://10.0.0.120:9000

Adres endpointu MinIO (czyli Twój serwer S3-kompatybilny).
To nie AWS — ale S3 API jest zgodne, więc Spark używa s3a://.

--conf spark.hadoop.fs.s3a.access.key=admin
--conf spark.hadoop.fs.s3a.secret.key=xxx

Klucze dostępu do MinIO (jak AWS AccessKey i SecretKey).

⚠️ W środowiskach produkcyjnych lepiej przechowywać je w:

Kubernetes Secrets,
albo używać IAM Role (np. w EKS).

--conf spark.hadoop.fs.s3a.path.style.access=true

MinIO (i starsze S3 implementacje) wymagają path-style access, np.:

http://10.0.0.120:9000/bucket/file.txt

--conf spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem

Ustawia klasę implementującą system plików s3a://.
Bez tego Spark mógłby nie rozpoznać ścieżek s3a://.

s3a://scripts/test.py

To plik z kodem Pythona, który Spark ma uruchomić. Znajduje się w MinIO w bucket’cie scripts.