将自定义容器与 Dataproc Serverless for Spark 搭配使用

Dataproc Serverless for Spark 在 Docker 容器中运行工作负载。容器为工作负载的驱动程序和执行程序进程提供运行时环境。默认情况下,Dataproc Serverless for Spark 使用容器映像,其中包含与运行时发布版本关联的默认 Spark、Java、Python 和 R 软件包。借助 Dataproc Serverless for Spark 批处理 API,您可以使用自定义容器映像而不是默认映像。通常,自定义容器映像会添加默认容器映像未提供的 Spark 工作负载 Java 或 Python 依赖项。重要提示:请勿在自定义容器映像中包含 Spark;Dataproc Serverless for Spark 将在运行时将 Spark 装载到容器中。

使用自定义容器映像提交 Spark 批处理工作负载

gcloud

提交 Spark 批处理工作负载时,使用带有 --container-image 标志的 gcloud dataproc batches submit spark 命令指定自定义容器映像。

gcloud dataproc batches submit spark \
    --container-image=custom-image, for example, "gcr.io/my-project-id/my-image:1.0.1" \
    --region=region \
    --jars=path to user workload jar located in Cloud Storage or included in the custom container \
    --class=The fully qualified name of a class in the jar file, such as org.apache.spark.examples.SparkPi \
    -- add any workload arguments here

注意

  • Custom-image:使用以下 Container Registry 映像命名格式指定自定义容器映像:{hostname}/{project-id}/{image}:{tag},例如“gcr.io/my-project-id/my-image:1.0.1”。注意:您必须在 Container RegistryArtifact Registry 上托管自定义容器映像。 (Dataproc 无服务器无法从其他注册表提取容器)。
  • --jars:指定包含在自定义容器映像中或 Cloud Storage 中的用户工作负载的路径,例如 /opt/spark/jars/spark-examples.jargs://my-bucket/spark/jars/spark-examples.jar
  • 其他批处理命令选项:您可以添加其他可选的批处理命令标志,例如使用永久性历史记录服务器 (PHS)。注意:PHS 必须位于运行批量工作负载的区域。
  • 工作负载参数:您可以通过在命令末尾添加“--”和工作负载参数来添加任何工作负载参数。

REST

自定义容器映像通过 RuntimeConfig.containerImage 字段作为 batches.create API 请求的一部分提供。

以下示例介绍如何使用自定义容器通过 Dataproc Spark for Spark batches.create API 提交批处理工作负载。

在使用任何请求数据之前,请先进行以下替换:

  • project-id:Google Cloud 项目 ID
  • region区域
  • custom-container-image:使用以下 Container Registry 映像命名格式指定自定义容器映像:{hostname}/{project-id}/{image}:{tag},例如“gcr.io/my-project-id/my-image:1.0.1”。 注意:您必须在 Container Registry Artifact Registry 上托管自定义容器。 (Dataproc 无服务器无法从其他注册表提取容器)。
  • jar-uri:指定包含在自定义容器映像中或位于 Cloud Storage 中的工作负载 jar 的路径,例如“/opt/spark/jars/spark-examples.jar” 或“gs:///spark/jars/spark-examples.jar”。
  • class:jar 文件中的类的完全限定名称,例如“org.apache.spark.examples.SparkPi”。
  • 其他选项:您可以使用其他批量工作负载资源字段,例如使用 sparkBatch.args 字段将参数传递给工作负载(如需了解详情,请参阅 Batch 资源文档)。如需使用永久性历史记录服务器 (PHS),请参阅设置永久性历史记录服务器。注意:PHS 必须位于运行批量工作负载的区域。

HTTP 方法和网址:

POST https://dataproc.googleapis.com/v1/projects/project-id/locations/region/batches

请求 JSON 正文:

{
  "runtimeConfig":{
    "containerImage":"custom-container-image
  },
  "sparkBatch":{
    "jarFileUris":[
      "jar-uri"
    ],
    "mainClass":"class"
  }
}

如需发送您的请求,请展开以下选项之一:

您应该收到类似以下内容的 JSON 响应:

{
"name":"projects/project-id/locations/region/batches/batch-id",
  "uuid":",uuid",
  "createTime":"2021-07-22T17:03:46.393957Z",
  "runtimeConfig":{
    "containerImage":"gcr.io/my-project/my-image:1.0.1"
  },
  "sparkBatch":{
    "mainClass":"org.apache.spark.examples.SparkPi",
    "jarFileUris":[
      "/opt/spark/jars/spark-examples.jar"
    ]
  },
  "runtimeInfo":{
    "outputUri":"gs://dataproc-.../driveroutput"
  },
  "state":"SUCCEEDED",
  "stateTime":"2021-07-22T17:06:30.301789Z",
  "creator":"account-email-address",
  "runtimeConfig":{
    "properties":{
      "spark:spark.executor.instances":"2",
      "spark:spark.driver.cores":"2",
      "spark:spark.executor.cores":"2",
      "spark:spark.app.name":"projects/project-id/locations/region/batches/batch-id"
    }
  },
  "environmentConfig":{
    "peripheralsConfig":{
      "sparkHistoryServerConfig":{
      }
    }
  },
  "operation":"projects/project-id/regions/region/operation-id"
}

构建自定义容器映像

Dataproc Serverless for Spark 自定义容器映像是 Docker 映像。您可以使用构建 Docker 映像的工具来构建自定义容器映像,但映像必须满足一些条件才能与 Dataproc Serverless for Spark 兼容。以下部分介绍了这些条件。

操作系统

您可以选择任何操作系统映像作为自定义容器映像的基础映像。建议:首选默认的 Debian 11 映像(例如 debian:11-slim),因为它们已完成测试,可以避免兼容性问题。

实用程序

您必须在自定义容器映像中添加运行 Spark 所需的以下实用程序软件包:

  • procps
  • tini

如需从 Spark(Java 或 Scala)运行 XGBoost,您必须添加 libgomp1

容器用户

Dataproc Serverless for Spark 以 spark Linux 用户身份通过 1099 UID 和 1099 GID 来运行容器。自定义容器映像 Dockerfile 中设置的 USER 指令会在运行时被忽略。使用 UID 和 GID 获得文件系统权限。 例如,如果您将映像中位于 /opt/spark/jars/my-lib.jar 中的 jar 文件添加为工作负载依赖项,则必须向 spark 用户授予该文件的读取权限。

映像流式传输

Dataproc Serverless for Spark 通常将整个映像下载到磁盘,以启动需要自定义容器映像的工作负载。这可能意味着初始化时间出现延迟,尤其是对图片较大的客户而言。

您可以改用映像流式传输,这种方法根据需要拉取映像数据。这样,工作负载就可以启动,而无需等待整个映像下载完毕,从而可能缩短初始化时间。如需启用映像流式传输,您必须启用 Container Filesystem API。您还必须将容器映像存储在 Artifact Registry 中,并且 Artifact Registry 代码库必须与 Dataproc 工作负载位于同一区域,或者位于与运行工作负载的区域对应的多区域中。如果 Dataproc 不支持该映像,或者映像流式传输服务不可用,则我们的流式传输实现会下载整个映像。请注意,我们在映像流式传输方面不支持以下各项:

在这些情况下,Dataproc 会在启动工作负载之前拉取整个映像。

Spark

请勿在自定义容器映像中添加 Spark。在运行时,Dataproc Serverless for Spark 将 Spark 二进制文件和配置从主机装载到容器中:二进制文件装载到 /usr/lib/spark 目录,配置装载到 /etc/spark/conf 目录。Dataproc Serverless for Spark 在运行时会替换这些目录中的现有文件。

Java 运行时环境

请勿在自定义容器映像中添加自己的 Java 运行时环境 (JRE)。在运行时,Dataproc Serverless for Spark 将 OpenJDK 从主机装载到容器中。如果您在自定义容器映像中添加 JRE,它将被忽略。

Java 软件包

您可以在自定义容器映像中添加 jar 文件作为 Spark 工作负载依赖项,并且可以将 SPARK_EXTRA_CLASSPATH 环境变量设置为包含 jar 文件。Dataproc Serverless for Spark 会将环境变量值添加到 Spark JVM 进程的类路径中。建议:将 jar 放入/opt/spark/jars目录下并将 SPARK_EXTRA_CLASSPATH 设置为 /opt/spark/jars/*

您可以在自定义容器映像中添加工作负载 jar,然后在提交工作负载时使用本地路径(例如 file:///opt/spark/jars/my-spark-job.jar)引用该工作负载 JAR(如需查看示例,请参阅使用自定义容器映像提交 Spark 批量工作负载)。

Python 软件包

默认情况下,Dataproc Serverless for Spark 会在运行时将 Conda 从主机装载到容器中的 /opt/dataproc/conda 目录。PYSPARK_PYTHON 已设置为 /opt/dataproc/conda/bin/python。其基本目录 /opt/dataproc/conda/bin 包含在 PATH 中。

您可以将 Python 环境的软件包添加到自定义容器映像的其他目录中(例如在 /opt/conda 中),并将 PYSPARK_PYTHON 环境变量设置为 /opt/conda/bin/python

您的自定义容器映像可以包含 Python 环境以外的其他 Python 模块,例如具有实用函数的 Python 脚本。设置 PYTHONPATH 环境变量,以包含模块所在的目录。

R 环境

您可以使用以下选项之一在自定义容器映像中自定义 R 环境:- 使用 Conda 从 conda-forge 渠道管理和安装 R 软件包 - 为您的容器映像 Linux 操作系统添加 R 代码库,并通过 Linux 操作系统软件包管理器安装 R 软件包(请参阅 R 软件包索引

使用任一选项时,您必须将 R_HOME 环境变量设置为指向您的自定义 R 环境。例外:如果您使用 Conda 管理 R 环境并自定义 Python 环境,则无需设置 R_HOME 环境变量;系统会根据 PYSPARK_PYTHON 环境变量自动设置该变量。

自定义容器映像构建示例

Dockerfile

# Debian 11 is recommended.
FROM debian:11-slim

# Suppress interactive prompts
ENV DEBIAN_FRONTEND=noninteractive

# (Required) Install utilities required by Spark scripts.
RUN apt update && apt install -y procps tini libjemalloc2

# (Optiona) Install utilities required by XGBoost for Spark.
RUN apt install -y procps libgomp1

# Enable jemalloc2 as default memory allocator
ENV LD_PRELOAD=/usr/lib/x86_64-linux-gnu/libjemalloc.so.2

# (Optional) Add extra jars.
ENV SPARK_EXTRA_JARS_DIR=/opt/spark/jars/
ENV SPARK_EXTRA_CLASSPATH='/opt/spark/jars/*'
RUN mkdir -p "${SPARK_EXTRA_JARS_DIR}"
COPY spark-bigquery-with-dependencies_2.12-0.22.2.jar "${SPARK_EXTRA_JARS_DIR}"

# (Optional) Install and configure Miniconda3.
ENV CONDA_HOME=/opt/miniconda3
ENV PYSPARK_PYTHON=${CONDA_HOME}/bin/python
ENV PATH=${CONDA_HOME}/bin:${PATH}
COPY Miniconda3-py39_4.10.3-Linux-x86_64.sh .
RUN bash Miniconda3-py39_4.10.3-Linux-x86_64.sh -b -p /opt/miniconda3 \
  && ${CONDA_HOME}/bin/conda config --system --set always_yes True \
  && ${CONDA_HOME}/bin/conda config --system --set auto_update_conda False \
  && ${CONDA_HOME}/bin/conda config --system --prepend channels conda-forge \
  && ${CONDA_HOME}/bin/conda config --system --set channel_priority strict
# Packages ipython and ipykernel are required if using custom conda and want to
# use this container for running notebooks.
RUN ${CONDA_HOME}/bin/conda install ipython ipykernel

# (Optional) Install Conda packages.
#
# The following packages are installed in the default image, it is strongly
# recommended to include all of them.
#
# Use mamba to install packages quickly.
RUN ${CONDA_HOME}/bin/conda install mamba -n base -c conda-forge \
    && ${CONDA_HOME}/bin/mamba install \
      conda \
      cython \
      fastavro \
      fastparquet \
      gcsfs \
      google-cloud-bigquery-storage \
      google-cloud-bigquery[pandas] \
      google-cloud-bigtable \
      google-cloud-container \
      google-cloud-datacatalog \
      google-cloud-dataproc \
      google-cloud-datastore \
      google-cloud-language \
      google-cloud-logging \
      google-cloud-monitoring \
      google-cloud-pubsub \
      google-cloud-redis \
      google-cloud-spanner \
      google-cloud-speech \
      google-cloud-storage \
      google-cloud-texttospeech \
      google-cloud-translate \
      google-cloud-vision \
      koalas \
      matplotlib \
      nltk \
      numba \
      numpy \
      openblas \
      orc \
      pandas \
      pyarrow \
      pysal \
      pytables \
      python \
      regex \
      requests \
      rtree \
      scikit-image \
      scikit-learn \
      scipy \
      seaborn \
      sqlalchemy \
      sympy \
      virtualenv

# (Optional) Add extra Python modules.
ENV PYTHONPATH=/opt/python/packages
RUN mkdir -p "${PYTHONPATH}"
COPY test_util.py "${PYTHONPATH}"

# (Optional) Install R and R libraries.
RUN apt update \
  && apt install -y gnupg \
  && apt-key adv --no-tty \
      --keyserver "hkp://keyserver.ubuntu.com:80" \
      --recv-keys E19F5F87128899B192B1A2C2AD5F960A256A04AF \
  && echo "deb http://cloud.r-project.org/bin/linux/debian bullseye-cran40/" \
      >/etc/apt/sources.list.d/cran-r.list \
  && apt update \
  && apt install -y \
      libopenblas-base \
      libssl-dev \
      r-base \
      r-base-dev \
      r-recommended \
      r-cran-blob

ENV R_HOME=/usr/lib/R

# (Required) Create the 'spark' group/user.
# The GID and UID must be 1099. Home directory is required.
RUN groupadd -g 1099 spark
RUN useradd -u 1099 -g 1099 -d /home/spark -m spark
USER spark

构建命令

在 Dockerfile 目录中运行。

IMAGE=gcr.io/my-project/my-image:1.0.1

# Download the BigQuery connector.
gsutil cp \
  gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.22.2.jar .

# Download the Miniconda3 installer.
wget https://repo.anaconda.com/miniconda/Miniconda3-py39_4.10.3-Linux-x86_64.sh

# Python module example
cat >test_util.py <<EOF
def hello(name):
  print("hello {}".format(name))

def read_lines(path):
  with open(path) as f:
    return f.readlines()
EOF

# Build and push the image.
docker build -t "${IMAGE}" .
docker push "${IMAGE}"