Service con Spark

Service con Spark

Sulla piattaforma Alida è possibile inserire applicazioni Spark all'interno dei Service. Tuttavia, per sfruttare appieno il potenziale del framework nel calcolo distribuito, è necessario apportare alcuni accorgimenti e strutturare i Service in modo diverso.

L'esecuzione di un Service Spark si divide in tre fasi:

1. Avvio dell'applicazione Spark tramite un comando spark-submit
2. Creazione di un pod contenente lo spark driver ed esecuzione del programma Spark principale al suo interno
3. Creazione dei nodi executor per l'esecuzione dei task in modo distribuito

Questo modello di esecuzione implica lo sviluppo di due immagini: una relativa allo spark-client - che gestisce la richiesta dell'applicazione Spark nel cluster Kubernetes - e una relativa al programma Spark vero e proprio. La gestione dei nodi driver/executor sarà automaticamente gestita da Spark.

Di seguito verrà esaminato in dettaglio il processo di sviluppo delle due immagini, facendo riferimento alla versione 4.x.y di Spark.

Applicazione Spark

L'immagine dell'applicazione Spark contiene il codice Spark vero e proprio, insieme alla logica di business. Ad esempio, se si utilizza Java, l'immagine conterrà il file .jar dell'applicazione, mentre se si utilizza Python, conterrà i vari moduli.

Sarà quindi necessario creare un Dockerfile che includa il codice eseguibile dell'applicazione. Si consiglia di utilizzare un'immagine di base ufficiale di Apache Spark, come apache/spark:4.0.1

Le configurazioni relative alla gestione delle risorse del cluster, così come le configurazioni specifiche del caso d'uso, saranno fornite dallo spark-client al momento dell'esecuzione dello spark-submit. Tra queste rientrano i consueti parametri inseriti da Alida, come argomenti o variabili globali.

Non sarà necessario creare un metamodello, poiché questa immagine non corrisponderà a un blocco Service nel designer, ma sarà indicata come parametro nel comando submit dello spark-client. L'immagine in questione verrà eseguita come spark-driver.

Va sottolineato che, per l'utilizzo in cluster di questa immagine da parte di Spark, essa non deve contenere un entrypoint. Questo perché sarà lo spark-submit a stabilire la modalità di esecuzione dell'immagine. Inserire un entrypoint personalizzato potrebbe impedire la corretta esecuzione del job Spark in modo distribuito, riducendola a una dimensione atomica su un singolo nodo driver.

Non è inoltre necessario definire un metamodello. Sarà sufficiente registrare il client in Alida, che istanzierà l'applicazione direttamente a partire dall'immagine Docker indicata, aggiungendo eventuali parametri e configurazioni.

Spark Client

L'immagine client sottopone la richiesta dell'applicazione Spark al cluster Kubernetes. Ciò comporterà innanzitutto la creazione di un container corrispondente allo spark driver e successivamente di eventuali altri container per gli executor. Sarà Spark stesso a occuparsi della creazione/distruzione dei pod in questione.

Funge inoltre da ponte tra l'applicazione Spark e Alida. A questo livello, informazioni e dati specifici di Alida, come variabili globali o parametri configurati dall'utente, verranno passati all'applicazione Spark vera e propria, in modo che possano essere acceduti come un normale Service Alida, tramite un opportuno metamodello.

In particolare, l'immagine client:

1. Recupera la variabile globale TARGET, contenente le specifiche di affinity e tolerations, e costruisce dinamicamente un manifest YAML basato su tali specifiche, secondo il quale verranno istanziati i pod dell'applicazione.
2. Esegue il comando spark-submit, inoltrando eventuali parametri/variabili di interesse presenti nel contesto di esecuzione Alida.
3. Recupera il codice di uscita del programma driver per aggiornare lo stato di esecuzione.

Di seguito verranno esaminati in dettaglio i punti chiave delle due immagini descritte. La versione di Spark di riferimento è la 4.0.1.

Dettagli di sviluppo - Client

Il cuore dello spark client è il comando spark-submit, inserito in un file .sh che funge da entrypoint per l'immagine. Questo entrypoint.sh conterrà porzioni di codice simili alle seguenti.

#!/bin/bash

# Variabili con valori di default (sovrascrivibili con export prima dello script)
SPARK_MASTER="${SPARK_MASTER:-local[*]}"
SPARK_NAMESPACE="${NAMESPACE:-alida}"
SPARK_DEPLOY_MODE="${SPARK_DEPLOY_MODE:-client}"
SPARK_APP_NAME="${SPARK_APP_NAME:-sparkapp}"
SPARK_IMAGE="${SPARK_IMAGE:-dockerhub.alidalab.it/alida/restricted/services/spark-kmeans-example:1.0.1}"
SPARK_DRIVER_MEMORY="${SPARK_DRIVER_MEMORY:-4g}"
SPARK_DRIVER_CORES="${SPARK_DRIVER_CORES:-2}"
SPARK_EXECUTOR_MEMORY="${SPARK_EXECUTOR_MEMORY:-4g}"
SPARK_EXECUTOR_CORES="${SPARK_EXECUTOR_CORES:-2}"
SPARK_PULLSECRETS="${SPARK_PULLSECRETS:-alida-regcred}"

TARGET="${TARGET:-$(cat <<'EOF'
{
   "affinity":{
      "nodeAffinity":{
         "requiredDuringSchedulingIgnoredDuringExecution":{
            "nodeSelectorTerms":[
               {
                  "matchExpressions":[
                     {
                        "key":"kubernetes.io/role",
                        "operator":"In",
                        "values":[
                           "opt-worker"
                        ]
                     }
                  ]
               }
            ]
         }
      }
   },
   "tolerations":[
      {
         "effect":"NoSchedule",
         "key":"opt-worker",
         "operator":"Equal",
         "value":"true"
      },
      {
         "effect":"NoExecute",
         "key":"opt-worker",
         "operator":"Equal",
         "value":"true"
      }
   ]
}
EOF
)}"

Innanzitutto vengono lette le variabili globali, qui inizializzate con valori di default. In particolare, la variabile TARGET contiene dati in formato JSON per lo YAML del pod.

# Estrae affinity e tolerations in variabili d'ambiente
AFFINITY=$(echo "$TARGET" | jq -c '.affinity')
TOLERATIONS=$(echo "$TARGET" | jq -c '.tolerations')


echo "CREAZIONE DEL DESCRITTORE YAML DEL POD A PARTIRE DALLA VARIABILE TARGET..."

# Converte il JSON in YAML utilizzando yq
TOLERATIONS_YAML=$(echo "$TOLERATIONS" | yq eval -P -)
AFFINITY_YAML=$(echo "$AFFINITY" | yq eval -P -)

# Scrive il risultato in pod-template.yaml
cat <<EOF > pod-template.yaml
apiVersion: v1
kind: Pod
spec:
  tolerations: 
$(echo "$TOLERATIONS_YAML" | sed 's/^/    /')
  affinity:
$(echo "$AFFINITY_YAML" | sed 's/^/    /')
EOF

Viene quindi costruito dinamicamente un file pod-template.yaml, che sarà indicato al comando spark-submit.

echo "INVIO DELL'APPLICAZIONE SPARK..."

OUTPUT="$(/opt/spark/bin/spark-submit \
    --master k8s://kubernetes.default.svc \
    --deploy-mode cluster \
    --name "${SPARK_APP_NAME}" \
    --conf spark.kubernetes.namespace="${SPARK_NAMESPACE}" \
    --conf spark.kubernetes.container.image=dockerhub.alidalab.it/alida/restricted/services/spark-kmeans-example:1.0.0 \
    --conf spark.kubernetes.driver.limit.memory="${SPARK_DRIVER_MEMORY}" \
    --conf spark.kubernetes.driver.limits.memory="${SPARK_DRIVER_MEMORY}" \
    --conf spark.kubernetes.driver.request.cores="${SPARK_DRIVER_CORES}" \
    --conf spark.kubernetes.executor.limit.memory="${SPARK_EXECUTOR_MEMORY}" \
    --conf spark.kubernetes.executor.limits.memory="${SPARK_EXECUTOR_MEMORY}" \
    --conf spark.kubernetes.executor.request.cores="${SPARK_EXECUTOR_CORES}" \
    --conf spark.kubernetes.driver.podTemplateFile=pod-template.yaml \
    --conf spark.kubernetes.executor.podTemplateFile=pod-template.yaml \
    --conf spark.kubernetes.container.image.pullSecrets="${SPARK_PULLSECRETS}" \
    --conf spark.kubernetes.driverEnv.EXECUTION_ID="${EXECUTION_ID}" \
    --conf spark.kubernetes.driverEnv.HASH_TOKEN="${HASH_TOKEN}" \
    --conf spark.kubernetes.driverEnv.SERVICE_ID="${SERVICE_ID}" \
    --conf spark.kubernetes.container.image.pullPolicy=Always \
    --conf spark.driver.memory="${SPARK_DRIVER_MEMORY}" \
    --conf spark.kubernetes.authenticate.driver.serviceAccountName=argo-workflow \
    --conf spark.hadoop.fs.s3a.path.style.access=true \
    --conf spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem \
    --conf spark.serializer=org.apache.spark.serializer.JavaSerializer \
    --conf spark.hadoop.fs.s3a.connection.timeout=60000 \
    --conf spark.hadoop.fs.s3a.connection.establish.timeout=5000 \
    --conf spark.hadoop.fs.s3a.attempts.maximum=10 \
    --conf spark.hadoop.fs.s3a.paging.maximum=1000 \
    --conf spark.hadoop.fs.s3a.connection.maximum=200 \
    --conf spark.hadoop.fs.s3a.aws.credentials.provider=org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider \
    --conf spark.hadoop.fs.s3a.threads.keepalivetime=60000 \
    --conf spark.hadoop.fs.s3a.connection.ttl=300000 \
    --conf spark.hadoop.fs.s3a.multipart.purge.age=86400000 \
    --conf spark.hadoop.fs.s3a.assumed.role.session.duration=1800000 \
    --conf spark.driver.userClassPathFirst=true \
    --conf spark.executor.userClassPathFirst=true \
    --conf spark.driver.log.localDir=/tmp \
    --conf spark.driver.extraJavaOptions="-Ddriver.log.dir=/tmp -Dlog4j2.configurationFile=/opt/spark/conf/log4j2.properties" \
    local:///app/src/main/main.py $@ 2>&1)"

Il comando spark-submit, riportando le variabili globali di interesse lette o inoltrate all'esecuzione, viene racchiuso in una variabile.

Il parametro spark.kubernetes.container.image fa riferimento al tag dell'immagine dell'applicazione.

Le configurazioni spark.kubernetes.authenticate.driver.serviceAccountName=argo-workflow e --conf spark.kubernetes.container.image.pullSecrets="${SPARK_PULLSECRETS}" sono necessarie per accedere correttamente alle immagini nel cluster Alida e per operare sui pod.

Le configurazioni del tipo spark.hadoop.fs.s3a... sono necessarie per l'integrazione con un filesystem di tipo S3, nel nostro caso S3. Nei casi in cui viene assegnato un timestamp, si restituiscono semplicemente i valori di default convertiti in millisecondi. Le librerie Hadoop necessarie per l'integrazione, infatti, utilizzano il formato temporale 60s, 24h…, il che rende necessario sovrascriverle in millisecondi come visto nei paragrafi precedenti.

I parametri userClassPathFirst assicurano che, in caso di conflitti tra librerie, vengano privilegiate le librerie indicate dal programmatore.

Per quanto riguarda il recupero dei log del driver, le variabili spark.driver.log.localDir e spark.driver.extraJavaOptions=-Ddriver.log.dir=/tmp -Dlog4j2.configurationFile=/opt/spark/conf/log4j2.properties assicurano che lo Spark driver scriva i log in una directory indicata all'interno del container del driver (immagine dell'applicazione).

$@ alla fine aggiunge gli argomenti di Alida, rendendoli visibili nel job Spark.

Limitazioni e problemi noti

Come già accennato, esiste un disallineamento tra le librerie Hadoop e Spark 4 riguardo al formato dei valori temporali. Spark 4 non supporta più il parsing di valori temporali nel formato 60s, 24h…, che invece sono utilizzati dalle librerie Hadoop per l'integrazione con S3, rendendo necessario sovrascriverli in millisecondi come visto nei paragrafi precedenti.

Un'altra limitazione riguarda la gestione delle configurazioni S3: queste vengono impostate globalmente nella configurazione Spark, all'inizio dell'esecuzione, memorizzate in cache e poi propagate agli executor. Eventuali modifiche durante l'esecuzione del job non verranno rilevate se non a livello del driver.

Questo è particolarmente rilevante per la gestione delle credenziali, poiché non consente di connettersi a percorsi S3 con credenziali diverse. Si raccomanda quindi di predisporre opportunamente l'input e l'output, posizionandoli tutti in un'area con le stesse credenziali.

In Alida, in particolare, è necessario assicurarsi che i dataset di input possano essere ricondotti allo stesso bucket/percorso del datasource di default creato al momento della creazione del workflow.