你总是可以使用一个兼容的解释器来运行PEX文件。所以你可以尝试使用python env.pex来代替指定./env.pex的程序。这并不要求env.pex是可执行的。

最后我无法直接运行pex，但现在确实得到了一个变通方案，这是pants slack community中的一个用户建议的（谢谢！）
解决方法是在集群初始化脚本中将pex解压缩为venv。
将初始化脚本gsutil复制到gs://<PATH TO INIT SCRIPT>：

#!/bin/bash
set -exo pipefail
readonly PEX_ENV_FILE_URI=$(/usr/share/google/get_metadata_value attributes/PEX_ENV_FILE_URI || true)
readonly PEX_FILES_DIR="/pexfiles"
readonly PEX_ENV_DIR="/pexenvs"
function err() {
    echo "[$(date +'%Y-%m-%dT%H:%M:%S%z')]: $*" >&2
    exit 1
}
function install_pex_into_venv() {
    local -r pex_name=${PEX_ENV_FILE_URI##*/}
    local -r pex_file="${PEX_FILES_DIR}/${pex_name}"
    local -r pex_venv="${PEX_ENV_DIR}/${pex_name}"
    echo "Installing pex from ${pex_file} into venv ${pex_venv}..."
    gsutil cp "${PEX_ENV_FILE_URI}" "${pex_file}"
    PEX_TOOLS=1 python "${pex_file}" venv --compile "${pex_venv}"
}
function main() {
    if [[ -z "${PEX_ENV_FILE_URI}" ]]; then
        err "ERROR: Must specify PEX_ENV_FILE_URI metadata key"
    fi
    install_pex_into_venv
}
main

要启动群集并运行初始化脚本以将pex解压缩到群集上的venv中，请执行以下操作：

from google.cloud import dataproc_v1 as dataproc
def start_cluster(project_id: str, region: str, cluster_name: str):
    cluster_client = dataproc.ClusterControllerClient(...)
    operation = cluster_client.create_cluster(
        request={
            "project_id": project_id,
            "region": region,
            "cluster": {
                "project_id": project_id,
                "cluster_name": cluster_name,
                "config": {
                    "master_config": <CONFIG>,
                    "worker_config": <CONFIG>,
                    "initialization_actions": [
                        {
                            "executable_file": "gs://<PATH TO INIT SCRIPT>",
                        },
                    ],
                    "gce_cluster_config": {
                        "metadata": {"PEX_ENV_FILE_URI": "gs://<PATH>/env.pex"},
                    },
                },
            },
        }
    )

要启动作业并使用解压缩的pex venv运行pyspark作业，请执行以下操作：

def submit_job(project_id: str, region: str, cluster_name: str):
    job_client = dataproc.ClusterControllerClient(...)
    operation = job_client.submit_job_as_operation(
        request={
            "project_id": project_id,
            "region": region,
            "job": {
                "placement": {"cluster_name": cluster_name},
                "pyspark_job": {
                    "main_python_file_uri": "gs://<PATH>/main.py",
                    "properties": {
                        "spark.pyspark.python": "/pexenvs/env.pex/bin/python",
                    },
                },
            },
        }
    )

下面@megabits的答案是适合我的基于bash的工作流
1.将初始化脚本（从answer）复制到GCS，作为gs://BUCKET/pkg/cluster-env-init.bash
1.构建PEX，提供初始化脚本所需的--include-tools参数，例如

pex --include-tools -r requirements.txt -o env.pex

1.将PEX文件放入GCS

gsutil mv env.pex "gs://BUCKET/pkg/env.pex"

1.使用PEX文件创建群集以设置环境

gcloud dataproc clusters create your-cluster --region us-central1 \
  --initialization-actions="gs://BUCKET/pkg/cluster-env-init.bash" \
  --metadata "PEX_ENV_FILE_URI=gs://BUCKET/pkg/env.pex"

1.运行作业

gcloud dataproc jobs submit pyspark your-script.py \
  --cluster=your-cluster --region us-central1 \
  --properties spark.pyspark.python="/pexenvs/env.pex/bin/python"