在GCP Dataproc集群运行Dask Python脚本的方法咨询

别担心，新手阶段遇到这类配置问题太正常了！我来给你一步步拆解怎么在Dataproc的1主2从集群上搭建Dask环境，包括完整的初始化脚本示例，以及后续运行你Dask代码的方法。

一、Dask初始化脚本（适配1主2从集群） #

这个脚本会自动在主节点启动Dask调度器（Scheduler），在从节点启动Dask工作节点（Worker），同时安装好conda和Dask依赖。你可以把下面的内容保存为init-dask.sh：

#!/bin/bash

# 安装基础依赖并配置conda
apt-get update -y
apt-get install -y wget

# 下载并静默安装Miniconda
wget https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh -q
bash Miniconda3-latest-Linux-x86_64.sh -b -p /opt/conda

# 把conda路径加入系统环境变量，确保所有用户都能访问
echo 'export PATH=/opt/conda/bin:$PATH' >> /etc/profile.d/conda.sh
source /etc/profile.d/conda.sh

# 安装Dask及配套工具（distributed用于分布式调度，bokeh用于监控）
conda install -y dask distributed bokeh

# 获取当前节点的Dataproc角色（Master/Worker）
ROLE=$(/usr/share/google/get_metadata_value attributes/dataproc-role)

if [[ "${ROLE}" == "Master" ]]; then
  # 主节点启动Dask调度器，后台运行并记录日志
  nohup dask scheduler --port 8786 > /var/log/dask-scheduler.log 2>&1 &
  echo "Dask Scheduler started on Master node"
elif [[ "${ROLE}" == "Worker" ]]; then
  # 从节点自动识别主节点主机名，连接调度器启动工作节点
  MASTER_HOSTNAME=$(hostname -f | sed 's/-w-[0-9]*//')
  nohup dask worker tcp://${MASTER_HOSTNAME}:8786 > /var/log/dask-worker.log 2>&1 &
  echo "Dask Worker started, connected to Master: ${MASTER_HOSTNAME}"
fi

脚本关键点说明： #

• 用dataproc-role元数据自动区分主从节点，不用手动配置
• 主节点的调度器默认监听8786端口，日志存在/var/log/dask-scheduler.log
• 从节点通过主机名规则自动找到主节点（Dataproc从节点主机名格式是集群名-w-序号，去掉-w-序号就是主节点名）
• 用nohup让Dask服务后台运行，避免SSH会话关闭后进程终止

二、创建Dataproc集群并应用初始化脚本 #

首先把你的init-dask.sh上传到Google Cloud Storage（GCS）的一个存储桶里（如果没有桶可以先创建一个）。然后用gcloud命令创建集群：

gcloud dataproc clusters create my-dask-cluster \
    --region us-central1 \
    --zone us-central1-a \
    --master-machine-type n1-standard-2 \
    --worker-machine-type n1-standard-2 \
    --num-workers 2 \
    --initialization-actions gs://你的存储桶名称/init-dask.sh

替换命令里的你的存储桶名称为你实际的GCS桶名，机器类型和区域可以根据你的需求调整。

三、运行你的Dask Python脚本 #

方法1：手动连接主节点运行 #

1. 用SSH连接到主节点：

   gcloud compute ssh my-dask-cluster-m --zone us-central1-a
   

2. 把你的Dask脚本上传到主节点：

   gcloud compute scp /本地路径/你的dask脚本.py my-dask-cluster-m:/home/你的用户名/
   

3. 修改你的Dask脚本，开头加上连接调度器的代码：

   from dask.distributed import Client
   
   # 连接到主节点的Dask调度器
   client = Client("tcp://localhost:8786")
   print("Dask集群连接成功:", client)
   
   # 你的原有Dask代码...
   

4. 在主节点运行脚本：

   python /home/你的用户名/你的dask脚本.py
   

方法2：用Dataproc作业提交（自动化方式） #

把你的Dask脚本上传到GCS，然后用下面的命令提交作业：

gcloud dataproc jobs submit pyspark gs://你的存储桶名称/你的dask脚本.py \
    --cluster my-dask-cluster \
    --region us-central1

这个方法不需要手动SSH，Dataproc会自动在主节点执行你的脚本，只要脚本里正确连接到本地的调度器即可。

四、验证Dask集群状态 #

你可以通过Dask的监控页面查看集群运行情况：

1. 本地执行端口转发命令，把主节点的8787端口映射到本地：

   gcloud compute ssh my-dask-cluster-m --zone us-central1-a -- -L 8787:localhost:8787
   

2. 在本地浏览器打开http://localhost:8787，就能看到Dask的监控界面，查看工作节点状态、任务进度等。

如果遇到问题，可以查看主/从节点的日志文件：/var/log/dask-scheduler.log和/var/log/dask-worker.log，排查依赖安装或服务启动的问题。

内容的提问来源于stack exchange，提问作者Mojtaba