Cari lettori, buon pomeriggio. Oggi parleremo un po' di Apache Spark e delle sue prospettive di sviluppo.
Nel mondo moderno dei Big Data, Apache Spark è diventato di fatto lo standard per lo sviluppo di compiti di elaborazione batch. Inoltre, viene utilizzato per creare applicazioni di streaming che operano con la micro batch, elaborando e spedendo dati in piccole porzioni (Spark Structured Streaming). Tradizionalmente, faceva parte dell'ecosistema Hadoop, utilizzando YARN come gestore delle risorse (o, in alcuni casi, Apache Mesos). Entro il 2020, il suo utilizzo nel formato tradizionale per la maggior parte delle aziende è messo in discussione a causa della mancanza di distribuzioni Hadoop decenti: lo sviluppo di HDP e CDH è stato fermato, CDH è poco ottimizzato e ha un costo elevato, mentre altri fornitori di Hadoop hanno cessato di esistere o hanno un futuro incerto. Pertanto, il lancio di Apache Spark tramite Kubernetes sta suscitando un crescente interesse da parte delle comunità e delle grandi aziende: poiché è diventato lo standard per l'orchestrazione dei container e la gestione delle risorse in nuvole private e pubbliche, risolve il problema della pianificazione inefficace delle risorse dei compiti Spark su YARN e fornisce una piattaforma in continua evoluzione con diverse distribuzioni commerciali e open source per aziende di tutte le dimensioni. Inoltre, con l'aumento della popolarità, molti hanno già implementato un paio di istanze proprie e hanno acquisito esperienza nel suo utilizzo, il che facilita la transizione.
A partire dalla versione 2.3.0, Apache Spark ha ottenuto il supporto ufficiale per l'esecuzione di compiti nel cluster Kubernetes e oggi parleremo della maturità attuale di questo approccio, delle diverse opzioni per il suo utilizzo e delle insidie che si potrebbero incontrare durante l'implementazione.
Prima di tutto, esaminiamo il processo di sviluppo di compiti e applicazioni basati su Apache Spark e evidenziamo casi tipici in cui è necessario avviare un compito su un cluster Kubernetes. Per la preparazione di questo post, viene utilizzata la distribuzione OpenShift e saranno fornite le comandi rilevanti per il suo strumento della riga di comando (oc). Per altre distribuzioni Kubernetes, possono essere utilizzati i comandi corrispondenti dello strumento standard della riga di comando Kubernetes (kubectl) o le loro alternative (ad esempio, per oc adm policy).
La prima opzione di utilizzo — spark-submit
Nel processo di sviluppo di compiti e applicazioni, lo sviluppatore deve avviare compiti per il debug delle trasformazioni dei dati. Teoricamente, per questi scopi possono essere utilizzati dei mock, ma lo sviluppo con l'uso di istanze reali (sebbene di test) dei sistemi finali si è dimostrato in questo tipo di lavoro più veloce e di qualità superiore. Nel caso in cui stiamo eseguendo il debug su istanze reali dei sistemi finali, sono possibili due scenari operativi:
- lo sviluppatore avvia un compito Spark localmente in modalità standalone;
- lo sviluppatore avvia un compito Spark su un cluster Kubernetes in un ambiente di test.
La prima opzione ha diritto di esistere, ma comporta alcuni svantaggi:
- è necessario garantire a ciascuno sviluppatore l'accesso dalle loro postazioni a tutte le istanze necessarie dei sistemi finali;
- sul computer di lavoro è necessario un numero sufficiente di risorse per avviare il compito in sviluppo.
La seconda opzione è priva di questi svantaggi, poiché l'uso del cluster Kubernetes consente di allocare il pool di risorse necessarie per avviare compiti e garantire i necessari accessi alle istanze dei sistemi finali, offrendo flessibilmente l'accesso tramite un modello di ruoli di Kubernetes a tutti i membri del team di sviluppo. Scegliamola come la prima opzione di utilizzo — avviare compiti Spark dalla macchina locale dello sviluppatore su un cluster Kubernetes in un ambiente di test.
Parliamo più in dettaglio del processo di configurazione di Spark per l'avvio locale. Per iniziare a utilizzare Spark, deve essere installato:
mkdir /opt/spark
cd /opt/spark
wget http://mirror.linux-ia64.org/apache/spark/spark-2.4.5/spark-2.4.5.tgz
tar zxvf spark-2.4.5.tgz
rm -f spark-2.4.5.tgz
Raccogliamo i pacchetti necessari per lavorare con Kubernetes:
cd spark-2.4.5/
./build/mvn -Pkubernetes -DskipTests clean package
La compilazione completa richiede molto tempo, e per creare immagini Docker e avviarle sul cluster Kubernetes sono realmente necessari solo i file jar dalla directory «assembly/», pertanto è possibile compilare solo questo sottoprogetto:
./build/mvn -f ./assembly/pom.xml -Pkubernetes -DskipTests clean package
Per eseguire attività Spark in Kubernetes è necessario creare un'immagine Docker che sarà utilizzata come base. Sono possibili due approcci:
- L'immagine Docker creata include il codice eseguibile dell'attività Spark;
- L'immagine creata include solo Spark e le dipendenze necessarie, il codice eseguibile è archiviato remotamente (per esempio, in HDFS).
Iniziamo a costruire un'immagine Docker contenente un esempio di attività Spark. Per creare immagini Docker, Spark ha un utilità specifica chiamata «docker-image-tool». Esaminiamo la sua guida:
./bin/docker-image-tool.sh --help
Con essa è possibile creare immagini Docker e caricarle in registri remoti, ma per impostazione predefinita presenta alcuni svantaggi:
- crea necessariamente immediatamente 3 immagini Docker — per Spark, PySpark e R;
- non consente di specificare il nome dell'immagine.
Pertanto, utilizzeremo una versione modificata di questo strumento, riportata di seguito:
vi bin/docker-image-tool-upd.sh
#!/usr/bin/env bash
function error {
echo "$@" 1>&2
exit 1
}
if [ -z "${SPARK_HOME}" ]; then
SPARK_HOME="$(cd "`dirname "$0"`"/..; pwd)"
fi
. "${SPARK_HOME}/bin/load-spark-env.sh"
function image_ref {
local image="$1"
local add_repo="${2:-1}"
if [ $add_repo = 1 ] && [ -n "$REPO" ]; then
image="$REPO/$image"
fi
if [ -n "$TAG" ]; then
image="$image:$TAG"
fi
echo "$image"
}
function build {
local BUILD_ARGS
local IMG_PATH
if [ ! -f "$SPARK_HOME/RELEASE" ]; then
IMG_PATH=$BASEDOCKERFILE
BUILD_ARGS=(
${BUILD_PARAMS}
--build-arg
img_path=$IMG_PATH
--build-arg
datagram_jars=datagram/runtimelibs
--build-arg
spark_jars=assembly/target/scala-$SPARK_SCALA_VERSION/jars
)
else
IMG_PATH="kubernetes/dockerfiles"
BUILD_ARGS=(${BUILD_PARAMS})
fi
if [ -z "$IMG_PATH" ]; then
error "Cannot find docker image. This script must be run from a runnable distribution of Apache Spark."
fi
if [ -z "$IMAGE_REF" ]; then
error "Cannot find docker image reference. Please add -i arg."
fi
local BINDING_BUILD_ARGS=(
${BUILD_PARAMS}
--build-arg
base_img=$(image_ref $IMAGE_REF)
)
local BASEDOCKERFILE=${BASEDOCKERFILE:-"$IMG_PATH/spark/docker/Dockerfile"}
docker build $NOCACHEARG "${BUILD_ARGS[@]}"
-t $(image_ref $IMAGE_REF)
-f "$BASEDOCKERFILE" .
}
function push {
docker push "$(image_ref $IMAGE_REF)"
}
function usage {
cat <<EOF
Usage: $0 [options] [command]
Builds or pushes the built-in Spark Docker image.
Commands:
build Build image. Requires a repository address to be provided if the image will be
pushed to a different registry.
push Push a pre-built image to a registry. Requires a repository address to be provided.
Options:
-f file Dockerfile to build for JVM based Jobs. By default builds the Dockerfile shipped with Spark.
-p file Dockerfile to build for PySpark Jobs. Builds Python dependencies and ships with Spark.
-R file Dockerfile to build for SparkR Jobs. Builds R dependencies and ships with Spark.
-r repo Repository address.
-i name Image name to apply to the built image, or to identify the image to be pushed.
-t tag Tag to apply to the built image, or to identify the image to be pushed.
-m Use minikube's Docker daemon.
-n Build docker image with --no-cache
-b arg Build arg to build or push the image. For multiple build args, this option needs to
be used separately for each build arg.
Using minikube when building images will do so directly into minikube's Docker daemon.
There is no need to push the images into minikube in that case, they'll be automatically
available when running applications inside the minikube cluster.
Check the following documentation for more information on using the minikube Docker daemon:
https://kubernetes.io/docs/getting-started-guides/minikube/#reusing-the-docker-daemon
Examples:
- Build image in minikube with tag "testing"
$0 -m -t testing build
- Build and push image with tag "v2.3.0" to docker.io/myrepo
$0 -r docker.io/myrepo -t v2.3.0 build
$0 -r docker.io/myrepo -t v2.3.0 push
EOF
}
if [[ "$@" = *--help ]] || [[ "$@" = *-h ]]; then
usage
exit 0
fi
REPO=
TAG=
BASEDOCKERFILE=
NOCACHEARG=
BUILD_PARAMS=
IMAGE_REF=
while getopts f:mr:t:nb:i: option
do
case "${option}"
in
f) BASEDOCKERFILE=${OPTARG};;
r) REPO=${OPTARG};;
t) TAG=${OPTARG};;
n) NOCACHEARG="--no-cache";;
i) IMAGE_REF=${OPTARG};;
b) BUILD_PARAMS=${BUILD_PARAMS}" --build-arg "${OPTARG};;
esac
done
case "${@: -1}" in
build)
build
;;
push)
if [ -z "$REPO" ]; then
usage
exit 1
fi
push
;;
*)
usage
exit 1
;;
esac
Con essa, costruiamo l'immagine di base di Spark, contenente un'attività di test per calcolare il numero Pi utilizzando Spark (qui {docker-registry-url} — URL del tuo registro di immagini Docker, {repo} — nome del repository all'interno del registro, corrispondente al progetto in OpenShift, {image-name} — nome dell'immagine (se è utilizzato un partizionamento a tre livelli delle immagini, ad esempio, come nel registro integrato di immagini Red Hat OpenShift), {tag} — tag di questa versione dell'immagine):
./bin/docker-image-tool-upd.sh -f resource-managers/kubernetes/docker/src/main/dockerfiles/spark/Dockerfile -r {docker-registry-url}/{repo} -i {image-name} -t {tag} build
Autenticati nel cluster OKD utilizzando lo strumento a riga di comando (qui {OKD-API-URL} — URL API del cluster OKD):
oc login {OKD-API-URL}
Otteniamo il token dell'utente corrente per l'autenticazione nel Docker Registry:
oc whoami -t
Ci autentichiamo nel Docker Registry interno del cluster OKD (utilizzando come password il token ottenuto con il comando precedente):
docker login {docker-registry-url}
Carichiamo l'immagine Docker costruita nel Docker Registry OKD:
./bin/docker-image-tool-upd.sh -r {docker-registry-url}/{repo} -i {image-name} -t {tag} push
Verifichiamo che l'immagine raccolta sia disponibile in OKD. Per fare ciò, apriamo nel browser l'URL con l'elenco delle immagini del progetto corrispondente (qui {project} è il nome del progetto all'interno del cluster OpenShift, {OKD-WEBUI-URL} è l'URL della console Web di OpenShift) — https://{OKD-WEBUI-URL}/console/project/{project}/browse/images/{image-name}.
Per avviare i lavori deve essere creato un account di servizio con privilegi per avviare pod come root (discuteremo ulteriormente questo punto):
oc create sa spark -n {project}
oc adm policy add-scc-to-user anyuid -z spark -n {project}
Eseguiamo il comando spark-submit per pubblicare il lavoro Spark nel cluster OKD, specificando l'account di servizio creato e l'immagine Docker:
/opt/spark/bin/spark-submit --name spark-test --class org.apache.spark.examples.SparkPi --conf spark.executor.instances=3 --conf spark.kubernetes.authenticate.driver.serviceAccountName=spark --conf spark.kubernetes.namespace={project} --conf spark.submit.deployMode=cluster --conf spark.kubernetes.container.image={docker-registry-url}/{repo}/{image-name}:{tag} --conf spark.master=k8s://https://{OKD-API-URL} local:///opt/spark/examples/target/scala-2.11/jars/spark-examples_2.11-2.4.5.jar
Qui:
-name — il nome del lavoro, che parteciperà alla formazione dei nomi dei pod di Kubernetes;
-class — la classe del file eseguibile chiamata all'avvio del lavoro;
-conf — i parametri di configurazione di Spark;
spark.executor.instances — il numero di esecutori Spark da avviare;
spark.kubernetes.authenticate.driver.serviceAccountName — il nome dell'account di servizio Kubernetes utilizzato per l'avvio dei pod (per definire il contesto di sicurezza e le autorizzazioni durante l'interazione con l'API Kubernetes);
spark.kubernetes.namespace — lo spazio dei nomi Kubernetes in cui verranno avviati i pod del driver e degli esecutori;
spark.submit.deployMode — il modo di eseguire Spark (per il normale spark-submit si utilizza "cluster", per Spark Operator e versioni successive di Spark "client");
spark.kubernetes.container.image — l'immagine Docker utilizzata per avviare i pod;
spark.master — l'URL dell'API Kubernetes (specificato esternamente quindi l'accesso avviene dalla macchina locale);
local:// — il percorso del file eseguibile Spark all'interno dell'immagine Docker.
Passiamo al progetto OKD corrispondente e esaminiamo i pod creati — https://{OKD-WEBUI-URL}/console/project/{project}/browse/pods.
Per semplificare il processo di sviluppo può essere utilizzata un'altra opzione, in cui viene creato un'immagine di base Spark condivisa, utilizzata da tutti i lavori per l'avvio, e gli snapshot dei file eseguibili vengono pubblicati in uno storage esterno (ad esempio, Hadoop) e specificati durante la chiamata a spark-submit come link. In questo modo è possibile avviare diverse versioni di lavori Spark senza ricompilare le immagini Docker, utilizzando, ad esempio, WebHDFS per pubblicare le immagini. Inviamo una richiesta per la creazione di un file (qui {host} è l'host del servizio WebHDFS, {port} è la porta del servizio WebHDFS, {path-to-file-on-hdfs} è il percorso desiderato per il file su HDFS):
curl -i -X PUT "http://{host}:{port}/webhdfs/v1/{path-to-file-on-hdfs}?op=CREATE
In questo caso si riceverà una risposta del tipo (qui {location} è l'URL da utilizzare per il caricamento del file):
HTTP/1.1 307 TEMPORARY_REDIRECT
Location: {location}
Content-Length: 0
Carichiamo il file eseguibile Spark in HDFS (qui {path-to-local-file} è il percorso del file eseguibile Spark sull'host corrente):
curl -i -X PUT -T {path-to-local-file} "{location}"
Dopo di che possiamo eseguire spark-submit utilizzando il file Spark caricato su HDFS (qui {class-name} è il nome della classe da eseguire per completare il compito):
/opt/spark/bin/spark-submit --name spark-test --class {class-name} --conf spark.executor.instances=3 --conf spark.kubernetes.authenticate.driver.serviceAccountName=spark --conf spark.kubernetes.namespace={project} --conf spark.submit.deployMode=cluster --conf spark.kubernetes.container.image={docker-registry-url}/{repo}/{image-name}:{tag} --conf spark.master=k8s://https://{OKD-API-URL} hdfs://{host}:{port}/{path-to-file-on-hdfs}
Si deve notare che per accedere a HDFS e garantire il funzionamento del compito potrebbe essere necessario modificare il Dockerfile e lo script entrypoint.sh: aggiungere nel Dockerfile una direttiva per copiare le librerie dipendenti nella directory /opt/spark/jars e includere il file di configurazione HDFS in SPARK_CLASSPATH in entrypoint.sh.
Una seconda opzione è Apache Livy
Successivamente, quando il compito è stato sviluppato e si deve testare il risultato ottenuto, sorge la questione del suo avvio all'interno del processo CI/CD e del monitoraggio degli stati della sua esecuzione. Certamente, è possibile eseguirlo anche tramite una chiamata locale a spark-submit, ma questo complica l'infrastruttura CI/CD poiché richiede l'installazione e la configurazione di Spark sugli agenti/runner del server CI e la configurazione dell'accesso all'API Kubernetes. Per questo caso, l'implementazione scelta è l'utilizzo di Apache Livy come API REST per avviare i compiti Spark, collocata all'interno del cluster Kubernetes. Con esso, è possibile eseguire compiti Spark sul cluster Kubernetes utilizzando normali richieste cURL, il che è facilmente realizzabile su qualsiasi soluzione CI, mentre la sua collocazione all'interno del cluster Kubernetes risolve la questione dell'autenticazione durante l'interazione con l'API Kubernetes.
Sottolineiamo questa come una seconda opzione: eseguire compiti Spark all'interno del processo CI/CD su un cluster Kubernetes in un ambiente di test.
Un po' su Apache Livy: funziona come server HTTP, fornendo un'interfaccia Web e un'API RESTful, che consente di avviare remotamente spark-submit, passando i parametri necessari. Tradizionalmente è stato fornito con la distribuzione HDP, ma può anche essere implementato in OKD o in qualsiasi altra installazione di Kubernetes utilizzando il manifesto appropriato e un insieme di immagini Docker, per esempio questo — . Per il nostro caso è stata costruita un'immagine Docker simile, contenente Spark versione 2.4.5 dal seguente Dockerfile:
DA java:8-alpine
ENV SPARK_HOME=/opt/spark
ENV LIVY_HOME=/opt/livy
ENV HADOOP_CONF_DIR=/etc/hadoop/conf
ENV SPARK_USER=spark
WORKDIR /opt
RUN apk add --update openssl wget bash &&
wget -P /opt https://downloads.apache.org/spark/spark-2.4.5/spark-2.4.5-bin-hadoop2.7.tgz &&
tar xvzf spark-2.4.5-bin-hadoop2.7.tgz &&
rm spark-2.4.5-bin-hadoop2.7.tgz &&
ln -s /opt/spark-2.4.5-bin-hadoop2.7 /opt/spark
RUN wget http://mirror.its.dal.ca/apache/incubator/livy/0.7.0-incubating/apache-livy-0.7.0-incubating-bin.zip &&
unzip apache-livy-0.7.0-incubating-bin.zip &&
rm apache-livy-0.7.0-incubating-bin.zip &&
ln -s /opt/apache-livy-0.7.0-incubating-bin /opt/livy &&
mkdir /var/log/livy &&
ln -s /var/log/livy /opt/livy/logs &&
cp /opt/livy/conf/log4j.properties.template /opt/livy/conf/log4j.properties
ADD livy.conf /opt/livy/conf
ADD spark-defaults.conf /opt/spark/conf/spark-defaults.conf
ADD entrypoint.sh /entrypoint.sh
ENV PATH="/opt/livy/bin:${PATH}"
EXPOSE 8998
ENTRYPOINT ["/entrypoint.sh"]
CMD ["livy-server"]
L'immagine creata può essere costruita e caricata nel tuo repository Docker esistente, ad esempio, il repository interno di OKD. Per la sua distribuzione viene utilizzato il seguente manifesto ({registry-url} — URL del registro delle immagini Docker, {image-name} — nome dell'immagine Docker, {tag} — tag dell'immagine Docker, {livy-url} — URL desiderato tramite il quale sarà accessibile il server Livy; il manifesto "Route" è applicato se come distribuzione Kubernetes viene utilizzato Red Hat OpenShift, altrimenti viene utilizzato il corrispondente manifesto Ingress o Service di tipo NodePort):
---
apiVersion: apps/v1
kind: Deployment
metadata:
labels:
component: livy
name: livy
spec:
progressDeadlineSeconds: 600
replicas: 1
revisionHistoryLimit: 10
selector:
matchLabels:
component: livy
strategy:
rollingUpdate:
maxSurge: 25%
maxUnavailable: 25%
type: RollingUpdate
template:
metadata:
creationTimestamp: null
labels:
component: livy
spec:
containers:
- command:
- livy-server
env:
- name: K8S_API_HOST
value: localhost
- name: SPARK_KUBERNETES_IMAGE
value: 'gnut3ll4/spark:v1.0.14'
image: '{registry-url}/{image-name}:{tag}'
imagePullPolicy: Always
name: livy
ports:
- containerPort: 8998
name: livy-rest
protocol: TCP
resources: {}
terminationMessagePath: /dev/termination-log
terminationMessagePolicy: File
volumeMounts:
- mountPath: /var/log/livy
name: livy-log
- mountPath: /opt/.livy-sessions/
name: livy-sessions
- mountPath: /opt/livy/conf/livy.conf
name: livy-config
subPath: livy.conf
- mountPath: /opt/spark/conf/spark-defaults.conf
name: spark-config
subPath: spark-defaults.conf
- command:
- /usr/local/bin/kubectl
- proxy
- '--port'
- '8443'
image: 'gnut3ll4/kubectl-sidecar:latest'
imagePullPolicy: Always
name: kubectl
ports:
- containerPort: 8443
name: k8s-api
protocol: TCP
resources: {}
terminationMessagePath: /dev/termination-log
terminationMessagePolicy: File
dnsPolicy: ClusterFirst
restartPolicy: Always
schedulerName: default-scheduler
securityContext: {}
serviceAccount: spark
serviceAccountName: spark
terminationGracePeriodSeconds: 30
volumes:
- emptyDir: {}
name: livy-log
- emptyDir: {}
name: livy-sessions
- configMap:
defaultMode: 420
items:
- key: livy.conf
path: livy.conf
name: livy-config
name: livy-config
- configMap:
defaultMode: 420
items:
- key: spark-defaults.conf
path: spark-defaults.conf
name: livy-config
name: spark-config
---
apiVersion: v1
kind: ConfigMap
metadata:
name: livy-config
data:
livy.conf: |-
livy.spark.deploy-mode=cluster
livy.file.local-dir-whitelist=/opt/.livy-sessions/
livy.spark.master=k8s://http://localhost:8443
livy.server.session.state-retain.sec = 8h
spark-defaults.conf: 'spark.kubernetes.container.image "gnut3ll4/spark:v1.0.14"'
---
apiVersion: v1
kind: Service
metadata:
labels:
app: livy
name: livy
spec:
ports:
- name: livy-rest
port: 8998
protocol: TCP
targetPort: 8998
selector:
component: livy
sessionAffinity: None
type: ClusterIP
---
apiVersion: route.openshift.io/v1
kind: Route
metadata:
labels:
app: livy
name: livy
spec:
host: {livy-url}
port:
targetPort: livy-rest
to:
kind: Service
name: livy
weight: 100
wildcardPolicy: None
Dopo l'applicazione e il successo nel lancio del pod, l'interfaccia grafica di Livy è disponibile al seguente link: http://{livy-url}/ui. Con Livy possiamo pubblicare il nostro compito Spark utilizzando una richiesta REST, ad esempio da Postman. Di seguito è fornito un esempio di collezione con le richieste (nel array «args» possono essere passati gli argomenti di configurazione con le variabili necessarie per l'esecuzione del compito lanciato):
{
"info": {
"_postman_id": "be135198-d2ff-47b6-a33e-0d27b9dba4c8",
"name": "Spark Livy",
"schema": "https://schema.getpostman.com/json/collection/v2.1.0/collection.json"
},
"item": [
{
"name": "1 Invia lavoro con jar",
"request": {
"method": "POST",
"header": [
{
"key": "Content-Type",
"value": "application/json"
}
],
"body": {
"mode": "raw",
"raw": "{nt"file": "local://opt/spark/examples/target/scala-2.11/jars/spark-examples_2.11-2.4.5.jar", nt"className": "org.apache.spark.examples.SparkPi",nt"numExecutors":1,nt"name": "spark-test-1",nt"conf": {ntt"spark.jars.ivy": "/tmp/.ivy",ntt"spark.kubernetes.authenticate.driver.serviceAccountName": "spark",ntt"spark.kubernetes.namespace": "{project}",ntt"spark.kubernetes.container.image": "{docker-registry-url}/{repo}/{image-name}:{tag}"nt}n}"
},
"url": {
"raw": "http://{livy-url}/batches",
"protocol": "http",
"host": [
"{livy-url}"
],
"path": [
"batches"
]
}
},
"response": []
},
{
"name": "2 Invia lavoro senza jar",
"request": {
"method": "POST",
"header": [
{
"key": "Content-Type",
"value": "application/json"
}
],
"body": {
"mode": "raw",
"raw": "{nt"file": "hdfs://{host}:{port}/{path-to-file-on-hdfs}", nt"className": "{class-name}",nt"numExecutors":1,nt"name": "spark-test-2",nt"proxyUser": "0",nt"conf": {ntt"spark.jars.ivy": "/tmp/.ivy",ntt"spark.kubernetes.authenticate.driver.serviceAccountName": "spark",ntt"spark.kubernetes.namespace": "{project}",ntt"spark.kubernetes.container.image": "{docker-registry-url}/{repo}/{image-name}:{tag}"nt},nt"args": [ntt"HADOOP_CONF_DIR=/opt/spark/hadoop-conf",ntt"MASTER=k8s://https://kubernetes.default.svc:8443"nt]n}"
},
"url": {
"raw": "http://{livy-url}/batches",
"protocol": "http",
"host": [
"{livy-url}"
],
"path": [
"batches"
]
}
},
"response": []
}
],
"event": [
{
"listen": "prerequest",
"script": {
"id": "41bea1d0-278c-40c9-ad42-bf2e6268897d",
"type": "text/javascript",
"exec": [
""
]
}
},
{
"listen": "test",
"script": {
"id": "3cdd7736-a885-4a2d-9668-bd75798f4560",
"type": "text/javascript",
"exec": [
""
]
}
}
],
"protocolProfileBehavior": {}
}
Eseguiremo la prima richiesta dalla collezione, andremo all'interfaccia OKD e verificheremo che il lavoro sia stato avviato correttamente — https://{OKD-WEBUI-URL}/console/project/{project}/browse/pods. In questo modo, nell'interfaccia Livy (http://{livy-url}/ui) apparirà una sessione, all'interno della quale possiamo monitorare l'andamento del lavoro e studiare i log della sessione tramite l'API di Livy o l'interfaccia grafica.
Ora mostreremo il meccanismo di funzionamento di Livy. Per fare ciò, esamineremo i registri del contenitore Livy all'interno del pod con il server Livy — https://{OKD-WEBUI-URL}/console/project/{project}/browse/pods/{livy-pod-name}?tab=logs. Da essi si evince che alla chiamata dell'API REST di Livy, nel contenitore denominato "livy", viene eseguito spark-submit, simile a quello che abbiamo utilizzato sopra (qui {livy-pod-name} è il nome del pod creato con il server Livy). Nella collezione è presente anche una seconda richiesta che consente di eseguire attività con un file eseguibile Spark situato in remoto tramite il server Livy.
La terza opzione di utilizzo — Spark Operator
Adesso, quando il compito è stato collaudato, sorge la questione del suo avvio regolare. Il metodo nativo per l'avvio regolare di attività nel cluster Kubernetes è l'entità CronJob e si può utilizzare, ma in questo momento sta guadagnando molta popolarità l'uso di operatori per gestire le applicazioni su Kubernetes e per Spark esiste un operatore abbastanza maturo, utilizzato anche nelle soluzioni di livello Enterprise (ad esempio, Lightbend FastData Platform). Raccomandiamo di utilizzarlo: l'attuale versione stabile di Spark (2.4.5) ha capacità molto limitate per la configurazione dell'avvio delle attività Spark in Kubernetes, mentre nella prossima versione principale (3.0.0) è prevista un supporto completo per Kubernetes, ma la data di uscita rimane sconosciuta. Spark Operator compensa questa mancanza, aggiungendo parametri di configurazione importanti (ad esempio, il montaggio di ConfigMap con la configurazione di accesso a Hadoop nei pod Spark) e la possibilità di avviare regolarmente il compito secondo un programma.
Lo evidenziamo come la terza opzione di utilizzo — l'avvio regolare delle attività Spark nel cluster Kubernetes in un ambiente produttivo.
Spark Operator ha codice sorgente aperto e viene sviluppato nell'ambito di Google Cloud Platform — . La sua installazione può essere effettuata in 3 modi:
- All'interno dell'installazione di Lightbend FastData Platform/Cloudflow;
- Utilizzando Helm:
helm repo add incubator http://storage.googleapis.com/kubernetes-charts-incubator helm install incubator/sparkoperator --namespace spark-operator - Utilizzando i manifesti dal repository ufficiale (https://github.com/GoogleCloudPlatform/spark-on-k8s-operator/tree/master/manifest). Va notato che Cloudflow include un operatore con versione API v1beta1. Se si utilizza questo tipo di installazione, le descrizioni dei manifesti delle applicazioni Spark devono basarsi su esempi da tag in Git con la corrispondente versione API, ad esempio, "v1beta1-0.9.0-2.4.0". È possibile visualizzare la versione dell'operatore nella descrizione CRD, presente nel dizionario "versions":
oc get crd sparkapplications.sparkoperator.k8s.io -o yaml
Se l'operatore è stato installato correttamente, nel progetto corrispondente apparirà un pod attivo con l'operatore Spark (ad esempio, cloudflow-fdp-sparkoperator nello spazio Cloudflow per l'installazione di Cloudflow) e apparirà il corrispondente tipo di risorse Kubernetes con il nome "sparkapplications". Per esaminare le applicazioni Spark disponibili, è possibile utilizzare il seguente comando:
oc get sparkapplications -n {project}
Per eseguire compiti utilizzando Spark Operator è necessario fare 3 cose:
- creare un'immagine Docker che includa tutte le librerie necessarie, oltre a file di configurazione ed eseguibili. Nello scenario target, questa è l'immagine creata nella fase CI/CD e testata su un cluster di test;
- pubblicare l'immagine Docker in un registro accessibile dal cluster Kubernetes;
- formulare un manifesto con tipo "SparkApplication" e descrizione del compito da eseguire. Esempi di manifesti sono disponibili nel repository ufficiale (ad esempio, ). È importante notare i punti cruciali riguardanti il manifesto:
- nel dizionario "apiVersion" deve essere specificata la versione API corrispondente a quella dell'operatore;
- nel dizionario "metadata.namespace" deve essere specificato lo spazio dei nomi in cui verrà eseguita l'applicazione;
- nel dizionario "spec.image" deve essere indicato l'indirizzo dell'immagine Docker creata nel registro disponibile;
- nel dizionario "spec.mainClass" deve essere specificato il nome della classe del compito Spark che deve essere eseguita all'avvio del processo;
- nel dizionario "spec.mainApplicationFile" deve essere fornito il percorso del file jar eseguibile;
- nel dizionario "spec.sparkVersion" deve essere specificata la versione di Spark utilizzata;
- nel dizionario "spec.driver.serviceAccount" deve essere указана l'account di servizio all'interno dello spazio dei nomi Kubernetes corrispondente, che sarà usato per eseguire l'applicazione;
- nel dizionario "spec.executor" deve essere specificato il numero di risorse dedicate all'applicazione;
- Nel dizionario «spec.volumeMounts» deve essere specificata la directory locale in cui verranno creati i file locali del task Spark.
Esempio di formazione di un manifesto (qui {spark-service-account} è l'account di servizio all'interno del cluster Kubernetes per l'esecuzione dei task Spark):
apiVersion: "sparkoperator.k8s.io/v1beta1"
kind: SparkApplication
metadata:
name: spark-pi
namespace: {project}
spec:
type: Scala
mode: cluster
image: "gcr.io/spark-operator/spark:v2.4.0"
imagePullPolicy: Always
mainClass: org.apache.spark.examples.SparkPi
mainApplicationFile: "local:////opt/spark/examples/jars/spark-examples_2.11-2.4.0.jar"
sparkVersion: "2.4.0"
restartPolicy:
type: Never
volumes:
- name: "test-volume"
hostPath:
path: "/tmp"
type: Directory
driver:
cores: 0.1
coreLimit: "200m"
memory: "512m"
labels:
version: 2.4.0
serviceAccount: {spark-service-account}
volumeMounts:
- name: "test-volume"
mountPath: "/tmp"
executor:
cores: 1
instances: 1
memory: "512m"
labels:
version: 2.4.0
volumeMounts:
- name: "test-volume"
mountPath: "/tmp"
In questo manifesto è specificato l'account di servizio, per il quale è necessario creare le necessarie binding di ruolo prima della pubblicazione del manifesto, che forniscono i permessi necessari per l'interazione dell'applicazione Spark con l'API Kubernetes (se necessario). Nel nostro caso, l'applicazione ha bisogno dei diritti per creare Pod. Creeremo la binding di ruolo necessaria:
oc adm policy add-role-to-user edit system:serviceaccount:{project}:{spark-service-account} -n {project}
Vale anche la pena notare che nella specifica di questo manifesto può essere specificato il parametro «hadoopConfigMap», che consente di indicare un ConfigMap con la configurazione di Hadoop senza la necessità di inserire in precedenza il relativo file nell'immagine Docker. È anche adatto per l'esecuzione programmata di task: utilizzando il parametro «schedule» può essere specificato il programma di esecuzione di questo task.
Dopo di che salviamo il nostro manifesto nel file spark-pi.yaml e lo applichiamo al nostro cluster Kubernetes:
oc apply -f spark-pi.yaml
In tal modo verrà creato un oggetto di tipo «sparkapplications»:
oc get sparkapplications -n {project}
> NAME AGE
> spark-pi 22h
In questo modo verrà creato un pod con l'applicazione, il cui stato sarà visibile nel «sparkapplications» creato. Può essere visualizzato con il comando seguente:
oc get sparkapplications spark-pi -o yaml -n {project}
Al termine del task, il POD passerà allo stato «Completed», che sarà aggiornato anche in «sparkapplications». I log dell'applicazione possono essere visualizzati nel browser o utilizzando il seguente comando (qui {sparkapplications-pod-name} è il nome del pod del task in esecuzione):
oc logs {sparkapplications-pod-name} -n {project}
La gestione delle attività di Spark può essere effettuata anche tramite l'utilità specializzata sparkctl. Per installarla, cloniamo il repository con il suo codice sorgente, installiamo Go e compiliamo questa utilità:
git clone https://github.com/GoogleCloudPlatform/spark-on-k8s-operator.git
cd spark-on-k8s-operator/
wget https://dl.google.com/go/go1.13.3.linux-amd64.tar.gz
tar -xzf go1.13.3.linux-amd64.tar.gz
sudo mv go /usr/local
mkdir $HOME/Projects
export GOROOT=/usr/local/go
export GOPATH=$HOME/Projects
export PATH=$GOPATH/bin:$GOROOT/bin:$PATH
go -version
cd sparkctl
go build -o sparkctl
sudo mv sparkctl /usr/local/bin
Esaminiamo l'elenco delle attività di Spark in esecuzione:
sparkctl list -n {project}
Creiamo una descrizione per l'attività di Spark:
vi spark-app.yaml
apiVersion: "sparkoperator.k8s.io/v1beta1"
kind: SparkApplication
metadata:
name: spark-pi
namespace: {project}
spec:
type: Scala
mode: cluster
image: "gcr.io/spark-operator/spark:v2.4.0"
imagePullPolicy: Always
mainClass: org.apache.spark.examples.SparkPi
mainApplicationFile: "local:////opt/spark/examples/jars/spark-examples_2.11-2.4.0.jar"
sparkVersion: "2.4.0"
restartPolicy:
type: Never
volumes:
- name: "test-volume"
hostPath:
path: "/tmp"
type: Directory
driver:
cores: 1
coreLimit: "1000m"
memory: "512m"
labels:
version: 2.4.0
serviceAccount: spark
volumeMounts:
- name: "test-volume"
mountPath: "/tmp"
executor:
cores: 1
instances: 1
memory: "512m"
labels:
version: 2.4.0
volumeMounts:
- name: "test-volume"
mountPath: "/tmp"
Avviamo l'attività descritta utilizzando sparkctl:
sparkctl create spark-app.yaml -n {project}
Esaminiamo l'elenco delle attività di Spark in esecuzione:
sparkctl list -n {project}
Esaminiamo l'elenco degli eventi dell'attività di Spark in esecuzione:
sparkctl event spark-pi -n {project} -f
Esaminiamo lo stato dell'attività di Spark in esecuzione:
sparkctl status spark-pi -n {project}
Infine, vorremmo considerare gli svantaggi riscontrati nell'uso dell'attuale versione stabile di Spark (2.4.5) in Kubernetes:
- Il primo e, probabilmente, principale svantaggio è l'assenza di Data Locality. Nonostante tutti i difetti, YARN aveva alcuni vantaggi, ad esempio il principio di consegna del codice ai dati (e non dei dati al codice). Grazie a questo, i task di Spark venivano eseguiti sui nodi dove si trovavano i dati coinvolti nei calcoli, riducendo notevolmente il tempo necessario per il trasferimento dei dati sulla rete. Con Kubernetes, ci troviamo nella necessità di spostare i dati attraverso la rete, dati che sono coinvolti nel lavoro del task. Se questi dati sono abbastanza grandi, il tempo di esecuzione del task può aumentare significativamente e potrebbe essere necessario un volume di spazio su disco considerevole, assegnato alle istanze del task Spark per il loro stoccaggio temporaneo. Questo svantaggio può essere ridotto attraverso l'uso di strumenti software specializzati che garantiscono la località dei dati in Kubernetes (ad esempio Alluxio), ma ciò implica praticamente la necessità di conservare una copia completa dei dati sui nodi del cluster Kubernetes.
- Un secondo importante svantaggio è la sicurezza. Di default, le funzionalità relative alla sicurezza per l'esecuzione dei task Spark sono disabilitate, l'utilizzo di Kerberos nella documentazione ufficiale non è trattato (anche se i parametri corrispondenti sono emersi nella versione 3.0.0, richiedendo ulteriori elaborazioni), e nella documentazione sulla sicurezza per l'utilizzo di Spark (https://spark.apache.org/docs/2.4.5/security.html) le sole opzioni di archiviazione delle chiavi sono YARN, Mesos e Standalone Cluster. Inoltre, l'utente sotto il quale vengono eseguiti i task Spark non può essere specificato direttamente: si imposta solo l'account di servizio sotto cui opererà il pod, e l'utente viene selezionato in base alle politiche di sicurezza configurate. Di conseguenza, si utilizza o l'utente root, che non è sicuro in un ambiente di produzione, oppure un utente con un UID casuale, il che è scomodo per la gestione dei diritti di accesso ai dati (risolvibile creando PodSecurityPolicies e legandole agli opportuni account di servizio). Attualmente, si affronta la situazione o posizionando tutti i file necessari direttamente nell'immagine Docker, o modificando lo script di avvio di Spark per utilizzare il meccanismo di archiviazione e recupero dei segreti adottato nella propria organizzazione.
- L'esecuzione di attività Spark tramite Kubernetes è ufficialmente ancora in fase sperimentale e potrebbero esserci cambiamenti significativi nei materiali utilizzati (file di configurazione, immagini Docker di base e script di avvio) in futuro. Infatti, durante la preparazione del materiale, sono state testate le versioni 2.3.0 e 2.4.5, e il comportamento era sostanzialmente diverso.
Aspettiamo aggiornamenti: recentemente è stata rilasciata una nuova versione di Spark (3.0.0), che ha portato cambiamenti significativi nel funzionamento di Spark su Kubernetes, pur mantenendo lo stato sperimentale del supporto per questo gestore di risorse. È possibile che i prossimi aggiornamenti consentano davvero di consigliare di abbandonare completamente YARN e di eseguire le attività Spark su Kubernetes, senza compromettere la sicurezza del sistema e senza la necessità di modifiche manuali ai componenti funzionali.
Fin.
Fonte: habr.com
