Dataproc主从节点Python版本不一致问题解决方案咨询

这个错误的核心很明确：你在主节点激活了Python3.x的虚拟环境运行PySpark驱动，但集群的worker节点还在使用默认的Python2.7，导致版本不匹配。要解决这个问题，我们需要把虚拟环境同步到所有worker节点，并让整个Spark集群统一使用虚拟环境里的Python。下面是分步解决方案：

1. 将主节点的虚拟环境同步到所有worker节点 #

你只在主节点创建了my-venv环境，worker节点并没有这个环境，所以第一步要把环境复制到每个worker：

步骤1：打包主节点的虚拟环境 #

在主节点执行以下命令打包环境：

tar -czf my-venv.tar.gz ~/anaconda3/envs/my-venv

步骤2：批量复制到所有worker节点 #

先获取集群的所有worker节点名称，然后循环复制和解压：

# 替换成你的集群名和区域
CLUSTER_NAME="your-cluster-name"
REGION="your-region"

# 获取所有worker节点列表
WORKERS=$(gcloud dataproc clusters describe $CLUSTER_NAME --region $REGION | grep 'name:' | grep worker | awk '{print $2}')

# 复制环境包到每个worker
for WORKER in $WORKERS; do
  gcloud compute scp my-venv.tar.gz $WORKER:~/ --zone $REGION
done

# 在每个worker节点解压环境包
for WORKER in $WORKERS; do
  gcloud dataproc ssh $WORKER --region $REGION --command "mkdir -p ~/anaconda3/envs && tar -xzf my-venv.tar.gz -C ~/anaconda3/envs/"
done

2. 配置PySpark统一使用虚拟环境的Python #

现在需要让Spark的驱动和worker都指向虚拟环境里的Python，有两种方式：

方式一：提交脚本时临时指定环境变量 #

每次提交PySpark脚本前，先激活虚拟环境，然后指定PYSPARK_PYTHON和PYSPARK_DRIVER_PYTHON：

conda activate my-venv
PYSPARK_PYTHON=~/anaconda3/envs/my-venv/bin/python PYSPARK_DRIVER_PYTHON=~/anaconda3/envs/my-venv/bin/python spark-submit --master yarn your_script.py

方式二：修改集群全局配置（永久生效） #

如果不想每次提交都指定变量，可以修改Spark的全局配置文件：

1. 在主节点和所有worker节点执行以下命令，添加环境变量到spark-env.sh：

echo 'export PYSPARK_PYTHON=~/anaconda3/envs/my-venv/bin/python' >> /etc/spark/conf/spark-env.sh
echo 'export PYSPARK_DRIVER_PYTHON=~/anaconda3/envs/my-venv/bin/python' >> /etc/spark/conf/spark-env.sh

2. 重启Spark服务使配置生效：

# 主节点重启历史服务器
sudo systemctl restart spark-history-server

# 每个worker节点重启worker服务
for WORKER in $WORKERS; do
  gcloud dataproc ssh $WORKER --region $REGION --command "sudo systemctl restart spark-worker"
done

3. 验证解决方案是否生效 #

可以在你的PySpark脚本中添加以下代码，验证Python版本和RDKit是否能正常加载：

import sys
import rdkit
from rdkit import Chem

# 打印版本信息
print(f"Driver Python version: {sys.version}")
print(f"RDKit version: {rdkit.__version__}")

# 测试RDKit功能（分布式执行）
def test_rdkit(smiles):
    mol = Chem.MolFromSmiles(smiles)
    return f"Parsed molecule: {mol}, Worker Python version: {sys.version}"

# 用RDD测试worker节点的执行
rdd = sc.parallelize(['C1=CC=CC=C1', 'CC(=O)O'])
results = rdd.map(test_rdkit).collect()
for res in results:
    print(res)

提交脚本后，如果输出的驱动和worker的Python版本一致，且RDKit能正常解析分子，说明问题已经解决。

优化建议：新建集群时用初始化动作自动配置 #

如果以后需要新建类似集群，可以写一个初始化动作脚本，自动创建虚拟环境、安装RDKit并配置Spark，避免手动同步的麻烦。示例脚本如下：

#!/bin/bash
# 创建虚拟环境并安装RDKit
conda create -y -n my-venv -c rdkit rdkit=2019.*
conda activate my-venv
conda install -y -c conda-forge rdkit

# 配置Spark使用虚拟环境的Python
echo 'export PYSPARK_PYTHON=/home/anaconda3/envs/my-venv/bin/python' >> /etc/spark/conf/spark-env.sh
echo 'export PYSPARK_DRIVER_PYTHON=/home/anaconda3/envs/my-venv/bin/python' >> /etc/spark/conf/spark-env.sh

# 重启Spark worker服务
sudo systemctl restart spark-worker

将脚本上传到GCS存储桶，然后创建集群时指定：

gcloud dataproc clusters create $CLUSTER_NAME \
    --region $REGION \
    --optional-components ANACONDA \
    --initialization-actions gs://your-bucket-path/init-script.sh

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