הפעלת Apache Spark ב-Kubernetes

קוראים יקרים, צהריים טובים. היום נדבר קצת על Apache Spark ועל סיכויי הפיתוח שלו.

בעולם המודרני של Big Data, Apache Spark הוא התקן דה פקטו לפיתוח משימות עיבוד נתונים אצווה. בנוסף, הוא משמש גם ליצירת אפליקציות סטרימינג שעובדות בקונספט ה-micro batch, עיבוד ומשלוח נתונים במנות קטנות (Spark Structured Streaming). ובאופן מסורתי זה היה חלק מהמחסנית הכוללת של Hadoop, תוך שימוש ב-YARN (או במקרים מסוימים Apache Mesos) כמנהל המשאבים. עד שנת 2020, השימוש בו במתכונתו המסורתית מוטל בספק עבור רוב החברות בגלל היעדר הפצות הגונות של Hadoop - הפיתוח של HDP ו-CDH נעצר, CDH אינו מפותח היטב ועלותו גבוהה, ושאר ספקי Hadoop או שהפסיק להתקיים או שיש לו עתיד עמום. לכן, השקת Apache Spark באמצעות Kubernetes מעוררת עניין הולך וגובר בקרב הקהילה וחברות גדולות - כשהיא הופכת לסטנדרט בתזמור קונטיינרים וניהול משאבים בעננים פרטיים וציבוריים, היא פותרת את הבעיה בתזמון משאבים לא נוח של משימות Spark ב-YARN ומספקת פלטפורמה מתפתחת בהתמדה עם הפצות מסחריות ופתוחות רבות לחברות בכל הגדלים והפסים. בנוסף, בעקבות הפופולריות, רובם כבר הצליחו לרכוש כמה מתקנים משלהם והגדילו את המומחיות שלהם בשימוש בו, מה שמפשט את המהלך.

החל מגרסה 2.3.0, Apache Spark רכשה תמיכה רשמית בהפעלת משימות באשכול Kubernetes והיום, נדבר על הבשלות הנוכחית של גישה זו, אפשרויות שונות לשימוש בה ועל המלכודות שייתקלו במהלך היישום.

קודם כל, בואו נסתכל על תהליך פיתוח משימות ויישומים המבוססים על Apache Spark ונדגיש מקרים טיפוסיים בהם צריך להריץ משימה על אשכול Kubernetes. בהכנת פוסט זה, OpenShift משמש כהפצה ויינתנו פקודות רלוונטיות לכלי השירות של שורת הפקודה (oc). עבור הפצות אחרות של Kubernetes, ניתן להשתמש בפקודות המתאימות משירות שורת הפקודה הסטנדרטי של Kubernetes (kubectl) או באנלוגים שלהן (לדוגמה, עבור מדיניות oc adm).

מקרה שימוש ראשון - spark-submit

במהלך הפיתוח של משימות ויישומים, המפתח צריך להפעיל משימות לניפוי באגים בטרנספורמציה של נתונים. תיאורטית, ניתן להשתמש בסטאב למטרות אלו, אך פיתוח הכולל מופעים אמיתיים (אם כי מבחן) של מערכות קצה הוכיח את עצמו כמהיר וטוב יותר במעמד זה של משימות. במקרה בו אנו מנפים באגים במופעים אמיתיים של מערכות קצה, שני תרחישים אפשריים:

• המפתח מריץ משימת Spark באופן מקומי במצב עצמאי;

  

• מפתח מריץ משימת Spark על אשכול Kubernetes בלולאת בדיקה.

  

לאפשרות הראשונה יש זכות קיום, אך טומנת בחובה מספר חסרונות:

• יש לספק לכל מפתח גישה ממקום העבודה לכל המופעים של מערכות הקצה שהוא צריך;
• נדרשת כמות מספקת של משאבים במכונה העובדת כדי להפעיל את המשימה המפותחת.

לאפשרות השנייה אין את החסרונות הללו, שכן השימוש באשכול Kubernetes מאפשר לך להקצות את מאגר המשאבים הדרוש להפעלת משימות ולספק לו את הגישה הדרושה למופעי קצה של מערכת, תוך מתן גישה אליו בצורה גמישה באמצעות מודל לחיקוי של Kubernetes עבור כל חברי צוות הפיתוח. בואו נדגיש את זה כמקרה השימוש הראשון - השקת משימות Spark ממחשב מפתח מקומי על אשכול Kubernetes במעגל בדיקה.

בואו נדבר יותר על תהליך הגדרת Spark להפעלה מקומית. כדי להתחיל להשתמש ב-Spark, עליך להתקין אותו:

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

אנו אוספים את החבילות הדרושות לעבודה עם Kubernetes:

cd spark-2.4.5/
./build/mvn -Pkubernetes -DskipTests clean package

בנייה מלאה לוקחת הרבה זמן, וכדי ליצור תמונות Docker ולהריץ אותן באשכול Kubernetes, אתה באמת צריך רק קבצי jar מספריית "assembly/", אז אתה יכול לבנות רק תת-פרויקט זה:

./build/mvn -f ./assembly/pom.xml -Pkubernetes -DskipTests clean package

כדי להפעיל עבודות Spark ב-Kubernetes, עליך ליצור תמונת Docker שתשמש כתמונת בסיס. יש כאן 2 גישות אפשריות:

• תמונת Docker שנוצרה כוללת את קוד המשימה Spark הניתן להפעלה;
• התמונה שנוצרה כוללת רק את Spark ואת התלות הנדרשת, קוד ההפעלה מתארח מרחוק (לדוגמה, ב-HDFS).

ראשית, בואו נבנה תמונת Docker המכילה דוגמה לבדיקה של משימת Spark. כדי ליצור תמונות Docker, ל-Spark יש כלי עזר שנקרא "docker-image-tool". בואו ללמוד את העזרה על זה:

./bin/docker-image-tool.sh --help

בעזרתו תוכלו ליצור תמונות Docker ולהעלות אותן לרישום מרוחקים, אך כברירת מחדל יש לה מספר חסרונות:

• בלי להיכשל יוצר 3 תמונות Docker בבת אחת - עבור Spark, PySpark ו-R;
• אינו מאפשר לך לציין שם תמונה.

לכן, אנו נשתמש בגרסה שונה של כלי השירות המופיע להלן:

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

בעזרתו, אנו מרכיבים תמונת Spark בסיסית המכילה משימת בדיקה לחישוב Pi באמצעות Spark (כאן {docker-registry-url} היא כתובת האתר של רישום התמונות של Docker, {repo} הוא שם המאגר בתוך הרישום, שמתאים לפרוייקט ב-OpenShift , {image-name} - שם התמונה (אם נעשה שימוש בהפרדה תלת-מפלסית של תמונות, למשל, כמו ברישום המשולב של Red Hat OpenShift תמונות), {tag} - תג של זה גרסה של התמונה):

./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

היכנס לאשכול OKD באמצעות כלי השירות המסוף (כאן {OKD-API-URL} היא כתובת ה-URL של OKD אשכול API):

oc login {OKD-API-URL}

בואו נשיג את האסימון של המשתמש הנוכחי להרשאה ב-Docker Registry:

oc whoami -t

היכנס ל-Docker Registry הפנימי של אשכול OKD (אנו משתמשים באסימון שהושג באמצעות הפקודה הקודמת כסיסמה):

docker login {docker-registry-url}

בואו נעלה את תמונת Docker המורכבת ל-Docker Registry OKD:

./bin/docker-image-tool-upd.sh -r {docker-registry-url}/{repo} -i {image-name} -t {tag} push

בואו נבדוק שהתמונה המורכבת זמינה ב-OKD. כדי לעשות זאת, פתח את כתובת האתר בדפדפן עם רשימת תמונות של הפרויקט המתאים (כאן {project} הוא שם הפרויקט בתוך אשכול OpenShift, {OKD-WEBUI-URL} הוא כתובת האתר של מסוף האינטרנט של OpenShift ) - https://{OKD-WEBUI-URL}/console /project/{project}/browse/images/{image-name}.

כדי להפעיל משימות, יש ליצור חשבון שירות עם הרשאות להפעיל פודים כ-root (נדון בנקודה זו מאוחר יותר):

oc create sa spark -n {project}
oc adm policy add-scc-to-user anyuid -z spark -n {project}

הבה נריץ את הפקודה spark-submit כדי לפרסם משימת Spark לאשכול OKD, תוך ציון חשבון השירות שנוצר ותמונת 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

Здесь:

—שם — שם המשימה שתשתתף בהיווצרות השם של תרמילי Kubernetes;

—class — מחלקה של קובץ ההפעלה, הנקראת כאשר המשימה מתחילה;

—conf — פרמטרי תצורת Spark;

spark.executor.instances - מספר המוציאים לפועל של Spark להפעלה;

spark.kubernetes.authenticate.driver.serviceAccountName - שם חשבון השירות Kubernetes המשמש בעת הפעלת פודים (כדי להגדיר את הקשר האבטחה והיכולות בעת אינטראקציה עם ה-API של Kubernetes);

spark.kubernetes.namespace - מרחב השמות של Kubernetes שבו יופעלו תרמילים של מנהלי התקנים ומבצעים;

spark.submit.deployMode - שיטת השקת Spark (עבור ה-Spark-submit סטנדרטי נעשה שימוש ב-"cluster", עבור Spark Operator וגרסאות מאוחרות יותר של Spark "client");

spark.kubernetes.container.image - תמונת Docker המשמשת להפעלת פודים;

spark.master - URL של Kubernetes API (חיצוני מצוין כך שהגישה מתרחשת מהמחשב המקומי);

local:// הוא הנתיב לקובץ ההפעלה של Spark בתוך תמונת Docker.

אנחנו הולכים לפרויקט OKD המתאים ולומדים את הפודים שנוצרו - https://{OKD-WEBUI-URL}/console/project/{project}/browse/pods.

כדי לפשט את תהליך הפיתוח, ניתן להשתמש באפשרות נוספת, בה נוצרת תמונת בסיס משותפת של Spark, המשמשת את כל המשימות להפעלה, ותמונות מצב של קבצי הפעלה מתפרסמות לאחסון חיצוני (לדוגמה, Hadoop) ומצוינות בעת קריאה spark-submit כקישור. במקרה זה, אתה יכול להריץ גרסאות שונות של משימות Spark מבלי לבנות מחדש תמונות Docker, באמצעות, למשל, WebHDFS כדי לפרסם תמונות. אנו שולחים בקשה ליצירת קובץ (כאן {host} הוא המארח של שירות WebHDFS, {port} הוא היציאה של שירות WebHDFS, {path-to-file-on-hdfs} הוא הנתיב הרצוי לקובץ ב-HDFS):

curl -i -X PUT "http://{host}:{port}/webhdfs/v1/{path-to-file-on-hdfs}?op=CREATE

תקבל תגובה כזו (הנה {location} היא כתובת האתר שיש להשתמש בה כדי להוריד את הקובץ):

HTTP/1.1 307 TEMPORARY_REDIRECT
Location: {location}
Content-Length: 0

טען את קובץ ההפעלה של Spark לתוך HDFS (כאן {path-to-local-file} הוא הנתיב לקובץ ההפעלה של Spark במארח הנוכחי):

curl -i -X PUT -T {path-to-local-file} "{location}"

לאחר מכן, נוכל לבצע spark-submit באמצעות קובץ ה-Spark שהועלה ל-HDFS (כאן {class-name} הוא שם המחלקה שצריך להפעיל כדי להשלים את המשימה):

/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}

יש לציין שכדי לגשת ל-HDFS ולהבטיח שהמשימה עובדת, ייתכן שיהיה עליך לשנות את ה- Dockerfile ואת הסקריפט entrypoint.sh - הוסף הנחיה ל- Dockerfile להעתקת ספריות תלויות לספריית /opt/spark/jars ו כלול את קובץ התצורה HDFS ב-SPARK_CLASSPATH בנקודת הכניסה. sh.

מקרה שימוש שני - Apache Livy

יתרה מכך, כאשר מפתחים משימה ויש לבדוק את התוצאה, עולה השאלה של השקתה כחלק מתהליך ה-CI/CD ומעקב אחר מצב ביצועה. כמובן, אתה יכול להפעיל אותו באמצעות קריאה מקומית של spark-submit, אבל זה מסבך את תשתית ה-CI/CD מכיוון שהיא דורשת התקנה והגדרה של Spark על סוכני/רצים של שרת CI והגדרת גישה ל-Kubernetes API. במקרה זה, הטמעת היעד בחרה להשתמש ב- Apache Livy בתור REST API להפעלת משימות Spark המתארחות בתוך אשכול Kubernetes. בעזרתו תוכלו להריץ משימות Spark על אשכול Kubernetes באמצעות בקשות cURL רגילות, אשר מיושמות בקלות בהתבסס על כל פתרון CI, והמיקום שלה בתוך אשכול Kubernetes פותר את בעיית האימות בעת אינטראקציה עם Kubernetes API.

בואו נדגיש את זה כמקרה שימוש שני - הפעלת משימות Spark כחלק מתהליך CI/CD על אשכול Kubernetes בלולאת בדיקה.

קצת על Apache Livy - הוא עובד כשרת HTTP המספק ממשק אינטרנט ו- RESTful API המאפשר להפעיל מרחוק spark-submit על ידי העברת הפרמטרים הדרושים. באופן מסורתי הוא נשלח כחלק מהפצת HDP, אך ניתן גם לפרוס אותו ל-OKD או לכל התקנה אחרת של Kubernetes באמצעות המניפסט המתאים ומערכת של תמונות Docker, כמו זו - github.com/ttauveron/k8s-big-data-experiments/tree/master/livy-spark-2.3. במקרה שלנו, נבנתה תמונת Docker דומה, כולל Spark גרסה 2.4.5 מה-Dockerfile הבא:

FROM 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"]

ניתן לבנות ולהעלות את התמונה שנוצרה למאגר Docker הקיים שלך, כגון מאגר OKD הפנימי. כדי לפרוס אותו, השתמש במניפסט הבא ({registry-url} - כתובת האתר של רישום התמונות של Docker, {image-name} - שם תמונת Docker, {tag} - תג התמונה של Docker, {livy-url} - כתובת האתר הרצויה שבה השרת יהיה נגיש Livy; המניפסט "Route" משמש אם Red Hat OpenShift משמש כהפצה Kubernetes, אחרת נעשה שימוש במניפסט Ingress או Service המקביל מסוג 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

לאחר החלתו והשקת הפוד בהצלחה, הממשק הגרפי של Livy זמין בקישור: http://{livy-url}/ui. עם Livy, נוכל לפרסם את משימת ה-Spark שלנו באמצעות בקשת REST של, למשל, Postman. דוגמה לאוסף עם בקשות מוצגת להלן (אפשר להעביר ארגומנטים של תצורה עם משתנים הנחוצים לפעולת המשימה שהושקה במערך "args"):

{
    "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 Submit job with 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 Submit job without 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": {}
}

בוא נבצע את הבקשה הראשונה מהאוסף, נעבור לממשק OKD ונבדוק שהמשימה הושקה בהצלחה - https://{OKD-WEBUI-URL}/console/project/{project}/browse/pods. במקביל, יופיע הפעלה בממשק Livy (http://{livy-url}/ui), שבתוכו, באמצעות ה-Livy API או הממשק הגרפי, תוכלו לעקוב אחר התקדמות המשימה וללמוד את ההפעלה יומנים.

עכשיו בואו נראה איך ליבי עובד. לשם כך, בואו נבחן את היומנים של מיכל Livy בתוך הפוד עם שרת Livy - https://{OKD-WEBUI-URL}/console/project/{project}/browse/pods/{livy-pod-name }?tab=logs. מהם אנו יכולים לראות שכאשר קוראים ל-Livy REST API במיכל בשם "livy", מתבצעת spark-submit, בדומה לזה שהשתמשנו למעלה (כאן {livy-pod-name} הוא השם של הפוד שנוצר עם שרת Livy). האוסף מציג גם שאילתה שנייה המאפשרת לך להריץ משימות המארחות מרחוק קובץ הפעלה של Spark באמצעות שרת Livy.

מקרה שימוש שלישי - Spark Operator

כעת, לאחר שהמשימה נבדקה, עולה השאלה של הפעלתה באופן קבוע. הדרך המקורית להריץ משימות באופן קבוע באשכול Kubernetes היא הישות CronJob ואתה יכול להשתמש בה, אבל כרגע השימוש באופרטורים לניהול יישומים ב- Kubernetes הוא מאוד פופולרי ועבור Spark יש אופרטור די בוגר, שהוא גם משמש בפתרונות ברמת הארגון (לדוגמה, Lightbend FastData Platform). אנו ממליצים להשתמש בו - לגרסה היציבה הנוכחית של Spark (2.4.5) יש אפשרויות תצורה מוגבלות למדי להפעלת משימות Spark ב-Kubernetes, בעוד שהגרסה הגדולה הבאה (3.0.0) מצהירה על תמיכה מלאה ב-Kubernetes, אך תאריך השחרור שלה לא ידוע . Spark Operator מפצה על חסרון זה על ידי הוספת אפשרויות תצורה חשובות (לדוגמה, הרכבה של ConfigMap עם תצורת גישה ל-Hadoop ל-Spark pods) והיכולת להפעיל משימה מתוזמנת באופן קבוע.

בואו נדגיש את זה כמקרה שימוש שלישי - הפעלת משימות Spark באופן קבוע על אשכול Kubernetes בלולאת ייצור.

Spark Operator הוא קוד פתוח ומפותח בתוך Google Cloud Platform - github.com/GoogleCloudPlatform/spark-on-k8s-operator. התקנתו יכולה להתבצע ב-3 דרכים:

1. כחלק מהתקנת Lightbend FastData Platform/Cloudflow;
2. שימוש בהגה:

   helm repo add incubator http://storage.googleapis.com/kubernetes-charts-incubator
   helm install incubator/sparkoperator --namespace spark-operator
   	

3. שימוש במניפסטים מהמאגר הרשמי (https://github.com/GoogleCloudPlatform/spark-on-k8s-operator/tree/master/manifest). ראוי לציין את הדברים הבאים - Cloudflow כולל מפעיל עם גרסת API v1beta1. אם נעשה שימוש בסוג זה של התקנה, תיאורי המניפסט של יישום Spark צריכים להתבסס על תגיות לדוגמה ב-Git עם גרסת ה-API המתאימה, למשל, "v1beta1-0.9.0-2.4.0". ניתן למצוא את גרסת האופרטור בתיאור ה-CRD הכלול באופרטור במילון "גרסאות":

   oc get crd sparkapplications.sparkoperator.k8s.io -o yaml
   	

אם המפעיל מותקן כהלכה, פוד פעיל עם המפעיל של Spark יופיע בפרויקט המתאים (לדוגמה, cloudflow-fdp-sparkoperator בחלל Cloudflow עבור התקנת Cloudflow) ויופיע סוג משאב Kubernetes מתאים בשם "sparkapplications" . אתה יכול לחקור יישומי Spark זמינים עם הפקודה הבאה:

oc get sparkapplications -n {project}

כדי להפעיל משימות באמצעות Spark Operator, עליך לעשות 3 דברים:

• צור תמונת Docker הכוללת את כל הספריות הדרושות, כמו גם קבצי תצורה והפעלה. בתמונת המטרה, זוהי תמונה שנוצרה בשלב ה-CI/CD ונבדקה על אשכול בדיקה;
• פרסם תמונת Docker לרישום הנגיש מאשכול Kubernetes;
• ליצור מניפסט עם סוג "SparkApplication" ותיאור של המשימה שתושק. מניפסטים לדוגמה זמינים במאגר הרשמי (למשל. github.com/GoogleCloudPlatform/spark-on-k8s-operator/blob/v1beta1-0.9.0-2.4.0/examples/spark-pi.yaml). יש נקודות חשובות לשים לב למניפסט:
  1. המילון "apiVersion" חייב לציין את גרסת ה-API המתאימה לגרסת המפעיל;
  2. המילון "metadata.namespace" חייב לציין את מרחב השמות שבו תופעל האפליקציה;
  3. המילון "spec.image" חייב להכיל את הכתובת של תמונת Docker שנוצרה ברישום נגיש;
  4. המילון "spec.mainClass" חייב להכיל את מחלקת המשימות Spark שיש להפעיל כאשר התהליך מתחיל;
  5. המילון "spec.mainApplicationFile" חייב להכיל את הנתיב לקובץ jar הניתן להפעלה;
  6. המילון "spec.sparkVersion" חייב לציין את הגרסה של Spark בשימוש;
  7. המילון "spec.driver.serviceAccount" חייב לציין את חשבון השירות בתוך מרחב השמות המתאים של Kubernetes שישמש להפעלת היישום;
  8. המילון "spec.executor" חייב לציין את מספר המשאבים שהוקצו לאפליקציה;
  9. המילון "spec.volumeMounts" חייב לציין את הספרייה המקומית שבה ייווצרו קבצי המשימות המקומיים של Spark.

דוגמה ליצירת מניפסט (כאן {spark-service-account} הוא חשבון שירות בתוך אשכול Kubernetes להפעלת משימות 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"

מניפסט זה מציין חשבון שירות שעבורו, לפני פרסום המניפסט, עליך ליצור את כריכות התפקידים הנחוצות המספקות את זכויות הגישה הדרושות ליישום Spark לאינטראקציה עם ה-API של Kubernetes (במידת הצורך). במקרה שלנו, האפליקציה זקוקה לזכויות ליצירת Pods. בואו ניצור את כריכת התפקידים הדרושה:

oc adm policy add-role-to-user edit system:serviceaccount:{project}:{spark-service-account} -n {project}

ראוי גם לציין שמפרט מניפסט זה עשוי לכלול פרמטר "hadoopConfigMap", המאפשר לך לציין ConfigMap עם תצורת Hadoop מבלי שתצטרך למקם תחילה את הקובץ המתאים בתמונת Docker. זה מתאים גם להפעלת משימות באופן קבוע - באמצעות פרמטר "לוח זמנים", ניתן לציין לוח זמנים להפעלת משימה נתונה.

לאחר מכן, אנו שומרים את המניפסט שלנו בקובץ spark-pi.yaml ומחילים אותו על אשכול Kubernetes שלנו:

oc apply -f spark-pi.yaml

פעולה זו תיצור אובייקט מסוג "sparkapplications":

oc get sparkapplications -n {project}
> NAME       AGE
> spark-pi   22h

במקרה זה, ייווצר פוד עם אפליקציה, שהסטטוס שלו יוצג ב-"sparkapplications". אתה יכול לראות את זה עם הפקודה הבאה:

oc get sparkapplications spark-pi -o yaml -n {project}

עם סיום המשימה, ה-POD יעבור למצב "הושלם", אשר יתעדכן גם ב-"sparkapplications". ניתן לצפות ביומני יישומים בדפדפן או באמצעות הפקודה הבאה (כאן {sparkapplications-pod-name} הוא שם הפוד של המשימה הפועלת):

oc logs {sparkapplications-pod-name} -n {project}

ניתן לנהל משימות Spark גם באמצעות כלי השירות המיוחד sparkctl. כדי להתקין אותו, שכפל את המאגר עם קוד המקור שלו, התקן את Go ובנה את כלי השירות הזה:

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

הבה נבחן את רשימת המשימות הרצות של Spark:

sparkctl list -n {project}

בואו ניצור תיאור עבור המשימה 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"

הבה נריץ את המשימה המתוארת באמצעות sparkctl:

sparkctl create spark-app.yaml -n {project}

הבה נבחן את רשימת המשימות הרצות של Spark:

sparkctl list -n {project}

הבה נבחן את רשימת האירועים של משימת Spark שהושקה:

sparkctl event spark-pi -n {project} -f

הבה נבחן את הסטטוס של משימת Spark הפועלת:

sparkctl status spark-pi -n {project}

לסיכום, ברצוני לשקול את החסרונות שהתגלו בשימוש בגרסה היציבה הנוכחית של Spark (2.4.5) ב-Kubernetes:

1. החיסרון הראשון ואולי העיקרי הוא היעדר Data Locality. למרות כל החסרונות של YARN, היו גם יתרונות לשימוש בו, למשל עקרון מסירת קוד לנתונים (ולא נתונים לקוד). הודות לו, בוצעו משימות Spark בצמתים שבהם נמצאו הנתונים המעורבים בחישובים, והזמן שנדרש להעברת נתונים ברשת הצטמצם משמעותית. בעת שימוש ב-Kubernetes, אנו מתמודדים עם הצורך להעביר נתונים המעורבים במשימה ברחבי הרשת. אם הם גדולים מספיק, זמן ביצוע המשימות יכול להגדיל באופן משמעותי, וגם לדרוש כמות די גדולה של שטח דיסק שהוקצה למופעי משימות Spark עבור האחסון הזמני שלהם. ניתן לצמצם את החיסרון הזה על ידי שימוש בתוכנה מיוחדת המבטיחה את מקומיות הנתונים ב-Kubernetes (לדוגמה, Alluxio), אך המשמעות היא למעשה הצורך לאחסן עותק שלם של הנתונים בצמתים של אשכול Kubernetes.
2. החיסרון החשוב השני הוא הביטחון. כברירת מחדל, תכונות הקשורות לאבטחה לגבי הפעלת משימות Spark מושבתות, השימוש ב-Kerberos אינו מכוסה בתיעוד הרשמי (אם כי האפשרויות המתאימות הוצגו בגרסה 3.0.0, שתדרוש עבודה נוספת), ותיעוד האבטחה עבור באמצעות Spark (https://spark.apache.org/docs/2.4.5/security.html) רק YARN, Mesos ו-Standalone Cluster מופיעים כחנויות מפתח. יחד עם זאת, לא ניתן לציין ישירות את המשתמש שתחתיו מופעלות משימות Spark - אנו מציינים רק את חשבון השירות תחתיו הוא יעבוד, והמשתמש נבחר על סמך מדיניות האבטחה המוגדרת. בהקשר זה, נעשה שימוש במשתמש השורש, שאינו בטוח בסביבה פרודוקטיבית, או במשתמש עם UID אקראי, שאינו נוח בעת הפצת זכויות גישה לנתונים (ניתן לפתור זאת על ידי יצירת PodSecurityPolicies וקישורם ל- חשבונות שירות תואמים). נכון לעכשיו, הפתרון הוא למקם את כל הקבצים הדרושים ישירות לתמונת Docker, או לשנות את סקריפט ההשקה של Spark כדי להשתמש במנגנון לאחסון ואחזור סודות שאומצו בארגון שלך.
3. הפעלת עבודות Spark באמצעות Kubernetes עדיין במצב ניסיוני וייתכנו שינויים משמעותיים בחפצים שבהם נעשה שימוש (קובצי תצורה, תמונות בסיס Docker וסקריפטי השקה) בעתיד. ואכן, בעת הכנת החומר, נבדקו גרסאות 2.3.0 ו-2.4.5, ההתנהגות הייתה שונה משמעותית.

בוא נחכה לעדכונים - לאחרונה שוחררה גרסה חדשה של Spark (3.0.0), שהביאה שינויים משמעותיים בעבודה של Spark ב-Kubernetes, אך שמרה על הסטטוס הניסיוני של תמיכה במנהל משאבים זה. אולי העדכונים הבאים באמת יאפשרו להמליץ ​​באופן מלא על נטישת YARN והפעלת משימות Spark ב-Kubernetes ללא חשש לאבטחת המערכת שלכם וללא צורך בשינוי עצמאי של רכיבים פונקציונליים.

סְנַפִּיר.

מקור: www.habr.com