Kubernetes での Apache Spark の実行

読者の皆様、こんにちは。 今日は、Apache Spark とその開発の見通しについて少しお話します。

現代のビッグ データの世界では、Apache Spark がバッチ データ処理タスクを開発するための事実上の標準です。 さらに、マイクロ バッチの概念で動作し、データを少しずつ処理して送信するストリーミング アプリケーション (Spark Structured Streaming) の作成にも使用されます。 そして伝統的に、YARN (場合によっては Apache Mesos) をリソース マネージャーとして使用し、Hadoop スタック全体の一部として使用されてきました。 2020 年までに、まともな Hadoop ディストリビューションが不足しているため、ほとんどの企業にとって従来の形式での使用が疑問視されています。HDP と CDH の開発は停止しており、CDH は十分に開発されておらず、コストが高く、残りの Hadoop サプライヤーは存在しなくなったか、暗い未来があるかのどちらかです。 したがって、Kubernetes を使用した Apache Spark の立ち上げは、コミュニティや大企業の間で関心が高まっています。プライベート クラウドとパブリック クラウドにおけるコンテナ オーケストレーションとリソース管理の標準となり、YARN 上の Spark タスクの不便なリソース スケジューリングの問題を解決し、次のような機能を提供します。あらゆる規模や種類の企業向けに多くの商用ディストリビューションとオープン ディストリビューションを備え、着実に発展しているプラ​​ットフォームです。 さらに、人気を受けて、ほとんどの企業はすでに独自のインストールをいくつか取得し、その使用に関する専門知識を高めているため、移行が簡素化されています。

バージョン 2.3.0 以降、Apache Spark は Kubernetes クラスターでタスクを実行するための公式サポートを取得しました。今日は、このアプローチの現在の成熟度、その使用のためのさまざまなオプション、および実装中に遭遇する落とし穴について説明します。

まず最初に、Apache Spark に基づいてタスクとアプリケーションを開発するプロセスを見て、Kubernetes クラスターでタスクを実行する必要がある一般的なケースを取り上げます。 この投稿を準備する際には、OpenShift がディストリビューションとして使用され、そのコマンド ライン ユーティリティ (oc) に関連するコマンドが提供されます。 他の Kubernetes ディストリビューションの場合は、標準の Kubernetes コマンド ライン ユーティリティ (kubectl) またはその類似物 (たとえば、oc adm ポリシー) の対応するコマンドを使用できます。

最初の使用例 - スパーク送信

タスクとアプリケーションの開発中、開発者はタスクを実行してデータ変換をデバッグする必要があります。 理論的には、スタブはこれらの目的に使用できますが、このクラスのタスクでは、エンド システムの実際の (テストではあるが) インスタンスを参加させた開発の方が高速かつ優れていることが証明されています。 エンド システムの実際のインスタンスでデバッグする場合、次の XNUMX つのシナリオが考えられます。

• 開発者は Spark タスクをスタンドアロン モードでローカルで実行します。

  

• 開発者はテストループで Kubernetes クラスター上で Spark タスクを実行します。

  

最初のオプションには存在する権利がありますが、多くの欠点が伴います。

• 各開発者には、職場から必要なエンド システムのすべてのインスタンスへのアクセスが提供されなければなりません。
• 開発中のタスクを実行するには、作業マシン上に十分な量のリソースが必要です。

XNUMX 番目のオプションには、これらの欠点はありません。Kubernetes クラスターを使用すると、タスクの実行に必要なリソース プールを割り当て、エンド システム インスタンスへの必要なアクセスを提供できるため、Kubernetes ロール モデルを使用してそのリソース プールへのアクセスを柔軟に提供できます。開発チームのメンバー全員。 最初のユースケースとして、テストループ内の Kubernetes クラスター上のローカル開発者マシンから Spark タスクを起動することを強調しましょう。

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 クラスター上で実行するには、実際に必要なのは「assembly/」ディレクトリの jar ファイルのみであるため、このサブプロジェクトのみをビルドできます。

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

Kubernetes で Spark ジョブを実行するには、基本イメージとして使用する Docker イメージを作成する必要があります。 ここで考えられるアプローチは 2 つあります。

• 生成された Docker イメージには、実行可能な Spark タスク コードが含まれています。
• 作成されたイメージには Spark と必要な依存関係のみが含まれており、実行可能コードはリモート (HDFS など) でホストされます。

まず、Spark タスクのテスト例を含む Docker イメージを構築しましょう。 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 イメージ レジストリの URL、{repo} はレジストリ内のリポジトリの名前、 OpenShift のプロジェクトと一致するもの、 {image-name} - イメージの名前 (たとえば、Red Hat OpenShift イメージの統合レジストリのようにイメージの XNUMX レベルの分離が使用されている場合)、 {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} は OKD クラスター API URL です)。

oc login {OKD-API-URL}

Docker レジストリで現在のユーザーの認証トークンを取得してみましょう。

oc whoami -t

OKD クラスターの内部 Docker レジストリにログインします (前のコマンドを使用して取得したトークンをパスワードとして使用します)。

docker login {docker-registry-url}

アセンブルされた Docker イメージを Docker レジストリ OKD にアップロードしましょう。

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

組み立てられたイメージが OKD で利用できることを確認してみましょう。 これを行うには、ブラウザで URL を開き、対応するプロジェクトのイメージのリストを表示します (ここで、{project} は OpenShift クラスター内のプロジェクトの名前、{OKD-WEBUI-URL} は OpenShift Web コンソールの URL です) ) - https://{OKD-WEBUI-URL}/console /project/{プロジェクト}/browse/images/{イメージ名}。

タスクを実行するには、ポッドを root として実行する権限を持つサービス アカウントを作成する必要があります (この点については後で説明します)。

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

作成したサービス アカウントと Docker イメージを指定して、spark-submit コマンドを実行して Spark タスクを OKD クラスターに公開しましょう。

 /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

ここに：

—name — Kubernetes ポッドの名前の形成に参加するタスクの名前。

—class — タスクの開始時に呼び出される実行可能ファイルのクラス。

—conf — Spark 設定パラメータ。

smile.executor.instances — 起動する Spark エグゼキューターの数。

smile.kubernetes.authenticate.driver.serviceAccountName - ポッドの起動時に使用される Kubernetes サービス アカウントの名前 (Kubernetes API と対話する際のセキュリティ コンテキストと機能を定義するため)。

smile.kubernetes.namespace — ドライバーおよびエグゼキューター ポッドが起動される Kubernetes 名前空間。

smile.submit.deployMode — Spark を起動するメソッド (標準の Spark-submit の「クラスター」、Spark Operator およびそれ以降のバージョンの Spark の「クライアント」の場合)。

smile.kubernetes.container.image - ポッドの起動に使用される Docker イメージ。

spark.master — Kubernetes API URL (ローカル マシンからアクセスが発生するように外部が指定されています)。

local:// は、Docker イメージ内の Spark 実行可能ファイルへのパスです。

対応する OKD プロジェクトに移動し、作成されたポッド (https://{OKD-WEBUI-URL}/console/project/{project}/browse/pods) を調べます。

開発プロセスを簡素化するために、別のオプションを使用できます。このオプションでは、Spark の共通の基本イメージが作成され、実行するすべてのタスクで使用され、実行可能ファイルのスナップショットが外部ストレージ (Hadoop など) に公開され、呼び出し時に指定されます。スパーク-リンクとして送信します。 この場合、WebHDFS を使用してイメージを公開するなど、Docker イメージを再構築せずに、さまざまなバージョンの Spark タスクを実行できます。 ファイルを作成するリクエストを送信します (ここで、{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} はファイルのダウンロードに使用する必要がある URL です)。

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

この後、HDFS にアップロードされた Spark ファイルを使用して、spark-submit を実行できます (ここで、{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 スクリプトを変更する必要がある場合があることに注意してください。依存するライブラリを /opt/spark/jars ディレクトリにコピーするためのディレクティブを Dockerfile に追加し、 HDFS 構成ファイルをエントリポイントの SPARK_CLASSPATH に含めます。

XNUMX 番目の使用例 - Apache Livy

さらに、タスクが開発され、その結果をテストする必要がある場合、CI/CD プロセスの一部としてタスクを起動し、その実行ステータスを追跡するかという問題が生じます。 もちろん、ローカルのspark-submit呼び出しを使用して実行することもできますが、CIサーバーのエージェント/ランナーにSparkをインストールして構成し、Kubernetes APIへのアクセスを設定する必要があるため、CI/CDインフラストラクチャが複雑になります。 この場合、ターゲット実装は、Kubernetes クラスター内でホストされる Spark タスクを実行するための REST API として Apache Livy を使用することを選択しました。 これを利用すると、通常の cURL リクエストを使用して Kubernetes クラスター上で Spark タスクを実行できます。これは、CI ソリューションに基づいて簡単に実装でき、Kubernetes クラスター内に配置することで、Kubernetes API と対話する際の認証の問題を解決します。

XNUMX 番目のユースケースとして、テスト ループ内の Kubernetes クラスター上の CI/CD プロセスの一部として Spark タスクを実行することを強調しましょう。

Apache Livy について少し説明します。Apache Livy は、Web インターフェイスと RESTful API を提供する HTTP サーバーとして機能します。これにより、必要なパラメーターを渡すことで、spark-submit をリモートで起動できるようになります。 従来、これは HDP ディストリビューションの一部として出荷されてきましたが、適切なマニフェストと次のような一連の Docker イメージを使用して、OKD またはその他の Kubernetes インストールにデプロイすることもできます。 github.com/ttauveron/k8s-big-data-experiments/tree/master/livy-spark-2.3。 この例では、次の Dockerfile からの Spark バージョン 2.4.5 を含む同様の Docker イメージが構築されました。

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

生成されたイメージはビルドして、内部 OKD リポジトリなどの既存の Docker リポジトリにアップロードできます。 これをデプロイするには、次のマニフェストを使用します ({registry-url} - Docker イメージ レジストリの URL、{image-name} - Docker イメージ名、{tag} - Docker イメージ タグ、{livy-url} - 希望の URLサーバーは Livy にアクセス可能になります。Red Hat OpenShift が Kubernetes ディストリビューションとして使用される場合は「Route」マニフェストが使用され、それ以外の場合は NodePort タイプの対応する Ingress マニフェストまたはサービス マニフェストが使用されます)。

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

これを適用してポッドを正常に起動すると、リンク http://{livy-url}/ui で Livy グラフィカル インターフェイスが利用可能になります。 Livy を使用すると、Postman などからの REST リクエストを使用して Spark タスクを公開できます。 リクエストを含むコレクションの例を以下に示します (起動されたタスクの操作に必要な変数を含む構成引数は、「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 サーバーを使用してポッド内の Livy コンテナのログを調べてみましょう - https://{OKD-WEBUI-URL}/console/project/{project}/browse/pods/{livy-pod-name }?tab=ログ。 これらから、「livy」という名前のコンテナーで Livy REST API を呼び出すと、上で使用したものと同様の Spark-Submit が実行されることがわかります (ここで、{livy-pod-name} は作成されたポッドの名前です) Livy サーバーを使用)。 このコレクションには、Livy サーバーを使用して Spark 実行可能ファイルをリモートでホストするタスクを実行できる XNUMX 番目のクエリも導入されています。

XNUMX 番目の使用例 - Spark オペレーター

タスクのテストが完了したので、定期的に実行するかどうかという問題が生じます。 Kubernetes クラスターでタスクを定期的に実行するネイティブな方法は CronJob エンティティであり、それを使用できますが、現時点では Kubernetes でアプリケーションを管理するためのオペレーターの使用が非常に人気があり、Spark にはかなり成熟したオペレーターがあります。エンタープライズレベルのソリューション (Lightbend FastData Platform など) で使用されます。 これを使用することをお勧めします。Spark の現在の安定バージョン (2.4.5) には、Kubernetes で Spark タスクを実行するための構成オプションがかなり制限されていますが、次のメジャー バージョン (3.0.0) では Kubernetes の完全なサポートが宣言されていますが、リリース日は不明のままです。 。 Spark Operator は、重要な構成オプション (たとえば、Hadoop アクセス構成を使用した ConfigMap を Spark ポッドにマウントする) と、定期的にスケジュールされたタスクを実行する機能を追加することで、この欠点を補います。

XNUMX 番目のユースケースとして、実稼働ループ内の Kubernetes クラスター上で Spark タスクを定期的に実行することを強調しましょう。

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 アプリケーションのマニフェストの説明は、適切な API バージョン (「v1beta1-0.9.0-2.4.0」など) を備えた Git のサンプル タグに基づいている必要があります。 演算子のバージョンは、「バージョン」辞書内の演算子に含まれる CRD の説明で確認できます。

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

オペレーターが正しくインストールされている場合、Spark オペレーターを含むアクティブなポッドが対応するプロジェクト (たとえば、Cloudflow インストールの Cloudflow スペース内の Cloudflow-fdp-sparkoperator) に表示され、「sparkapplications」という名前の対応する Kubernetes リソース タイプが表示されます。 。 次のコマンドを使用して、利用可能な Spark アプリケーションを探索できます。

oc get sparkapplications -n {project}

Spark Operator を使用してタスクを実行するには、次の 3 つのことを行う必要があります。

• 必要なすべてのライブラリ、構成ファイル、実行可能ファイルを含む Docker イメージを作成します。 ターゲットの図では、これは CI/CD 段階で作成され、テスト クラスターでテストされたイメージです。
• Kubernetes クラスターからアクセス可能なレジストリに Docker イメージを公開します。
• 「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. 実行可能 jar ファイルへのパスは「spec.mainApplicationFile」ディクショナリで指定する必要があります。
  6. 「spec.sparkVersion」辞書は、使用されている Spark のバージョンを示す必要があります。
  7. 「spec.driver.serviceAccount」ディクショナリでは、アプリケーションの実行に使用される、対応する Kubernetes 名前空間内のサービス アカウントを指定する必要があります。
  8. 「spec.executor」辞書は、アプリケーションに割り当てられたリソースの数を示す必要があります。
  9. 「spec.volumeMounts」ディクショナリでは、ローカル Spark タスク ファイルが作成されるローカル ディレクトリを指定する必要があります。

マニフェストの生成例 (ここで {spark-service-account} は、Spark タスクを実行するための Kubernetes クラスター内のサービス アカウントです)。

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"

このマニフェストは、マニフェストを公開する前に、(必要に応じて) Kubernetes API と対話するために Spark アプリケーションに必要なアクセス権を提供する必要なロール バインディングを作成する必要があるサービス アカウントを指定します。 この場合、アプリケーションにはポッドを作成する権限が必要です。 必要なロール バインディングを作成しましょう。

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

このマニフェスト仕様には「hadoopConfigMap」パラメータが含まれる場合があることにも注目してください。これにより、最初に対応するファイルを Docker イメージに配置しなくても、Hadoop 構成を使用して ConfigMap を指定できます。 また、タスクを定期的に実行するのにも適しています。「スケジュール」パラメータを使用すると、特定のタスクを実行するスケジュールを指定できます。

その後、マニフェストを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}

結論として、Kubernetes で Spark の現在の安定バージョン (2.4.5) を使用することで判明した欠点について考えてみたいと思います。

1. 最初の、そしておそらく主な欠点は、データの局所性が欠如していることです。 YARN にはすべての欠点がありますが、YARN を使用する利点もありました。たとえば、(データからコードではなく) コードをデータに配信する原則などです。 そのおかげで、計算に関係するデータが存在するノード上で Spark タスクが実行され、ネットワーク経由でデータを配信するのにかかる時間が大幅に短縮されました。 Kubernetes を使用する場合、タスクに関係するデータをネットワーク経由で移動する必要があります。 それらが十分に大きい場合、タスクの実行時間が大幅に増加する可能性があり、一時ストレージとして Spark タスク インスタンスに割り当てられたかなり大量のディスク領域も必要になります。 この欠点は、Kubernetes でのデータの局所性を保証する特殊なソフトウェア (Alluxio など) を使用することで軽減できますが、実際には、データの完全なコピーを Kubernetes クラスター ノードに保存する必要があることを意味します。
2. 3.0.0 番目の重要な欠点はセキュリティです。 デフォルトでは、Spark タスクの実行に関するセキュリティ関連の機能は無効になっており、Kerberos の使用は公式ドキュメントでカバーされていません (ただし、対応するオプションはバージョン 2.4.5 で導入されており、追加の作業が必要です)。 Spark (https://spark.apache.org/docs/XNUMX/security.html) を使用すると、YARN、Mesos、およびスタンドアロン クラスターのみがキー ストアとして表示されます。 同時に、Spark タスクを起動するユーザーを直接指定することはできません。タスクが動作するサービス アカウントのみを指定し、ユーザーは構成されたセキュリティ ポリシーに基づいて選択されます。 この点に関して、生産環境では安全ではない root ユーザーが使用されるか、ランダムな UID を持つユーザーが使用されます。これは、データへのアクセス権を配布するときに不便です (これは、PodSecurityPolicies を作成し、それらを対応するサービスアカウント)。 現時点での解決策は、必要なすべてのファイルを Docker イメージに直接配置するか、組織で採用されているシークレットを保存および取得するためのメカニズムを使用するように Spark 起動スクリプトを変更することです。
3. Kubernetes を使用した Spark ジョブの実行は、公式にはまだ実験モードであり、将来的に使用されるアーティファクト (構成ファイル、Docker ベース イメージ、および起動スクリプト) に大幅な変更が発生する可能性があります。 実際、資料を準備するときにバージョン 2.3.0 と 2.4.5 をテストしたところ、動作は大きく異なりました。

更新を待ちましょう - Spark の新しいバージョン (3.0.0) が最近リリースされました。これにより、Kubernetes 上の Spark の動作に大幅な変更が加えられましたが、このリソース マネージャーのサポートは実験的な状態が維持されました。 おそらく次のアップデートでは、システムのセキュリティを心配することなく、機能コンポーネントを個別に変更する必要もなく、YARN を放棄して Kubernetes 上で Spark タスクを実行することを完全に推奨できるようになるでしょう。

終わり

出所： habr.com