DGX服务器上基于Spark部署ComfyUI的AI绘画方案

1. 项目背景与核心需求

在深度学习与AI绘画领域，DGX系列工作站因其强大的GPU算力成为专业用户的首选设备。而ComfyUI作为一款基于节点式工作流的Stable Diffusion前端，相比传统WebUI具有更低显存占用和更高灵活性的优势。本教程将解决一个具体痛点：如何在DGX服务器上基于Spark分布式环境部署ComfyUI，实现多GPU协同的AI绘画工作流。

这个方案特别适合：

• 需要批量生成高清图像的内容工作室
• 进行AI绘画模型研发的算法团队
• 对工作流可定制性要求高的技术美术师

2. 环境准备与依赖检查

2.1 硬件配置确认

DGX系列服务器通常配备多块NVIDIA Tesla GPU（如A100/V100）。在开始前需要确认：

bash复制nvidia-smi  # 查看GPU型号和数量
lscpu       # 检查CPU和内存配置
df -h       # 确认存储空间

注意：ComfyUI对显存的要求取决于模型尺寸，建议单卡至少24GB显存用于运行SDXL模型

2.2 基础软件栈安装

1. 通过Spark官方脚本安装最新版Spark：

bash复制wget https://archive.apache.org/dist/spark/spark-3.5.0/spark-3.5.0-bin-hadoop3.tgz
tar -xzf spark-3.5.0-bin-hadoop3.tgz
sudo mv spark-3.5.0-bin-hadoop3 /opt/spark

2. 配置Python环境（建议使用Miniconda）：

bash复制wget https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh
bash Miniconda3-latest-Linux-x86_64.sh
conda create -n comfy python=3.10
conda activate comfy

3. ComfyUI核心部署流程

3.1 源码获取与初始化

使用Git克隆官方仓库并安装依赖：

bash复制git clone https://github.com/comfyanonymous/ComfyUI
cd ComfyUI
pip install torch torchvision torchaudio --extra-index-url https://download.pytorch.org/whl/cu118
pip install -r requirements.txt

3.2 Spark集成配置

1. 修改main.py添加Spark初始化代码：

python复制from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("ComfyUI-Spark") \
    .config("spark.executor.memory", "16g") \
    .config("spark.driver.memory", "8g") \
    .getOrCreate()

2. 创建分布式任务调度器：

python复制def distributed_inference(workflow_json):
    # 将工作流分发到各GPU节点执行的逻辑
    ...

4. 模型部署与优化技巧

4.1 模型文件管理

建议将模型文件存放在共享存储（如NFS）：

code复制models/
├── checkpoints/
├── loras/
├── vae/
└── ...

在extra_model_paths.yaml中配置：

yaml复制base_path: /mnt/nfs/models
checkpoints: checkpoints
loras: loras
vae: vae

4.2 多GPU负载均衡

通过修改execution.py实现动态负载分配：

python复制def queue_prompt(prompt):
    gpu_util = get_gpu_utilization()
    least_loaded = np.argmin(gpu_util)
    return send_to_device(prompt, device=f"cuda:{least_loaded}")

5. 生产环境部署方案

5.1 系统服务化配置

创建systemd服务文件/etc/systemd/system/comfyui.service：

ini复制[Unit]
Description=ComfyUI Service
After=network.target

[Service]
User=comfy
Group=comfy
WorkingDirectory=/opt/ComfyUI
ExecStart=/opt/miniconda3/envs/comfy/bin/python main.py
Restart=always

[Install]
WantedBy=multi-user.target

5.2 安全加固措施

1. 启用HTTPS：

bash复制openssl req -x509 -nodes -days 365 -newkey rsa:2048 \
  -keyout /etc/ssl/private/comfyui.key \
  -out /etc/ssl/certs/comfyui.crt

2. 配置Nginx反向代理：

nginx复制server {
    listen 443 ssl;
    server_name yourdomain.com;

    ssl_certificate /etc/ssl/certs/comfyui.crt;
    ssl_certificate_key /etc/ssl/private/comfyui.key;

    location / {
        proxy_pass http://localhost:8188;
        proxy_set_header Host $host;
    }
}

6. 性能调优实战

6.1 基准测试方法

使用自动化测试脚本评估吞吐量：

python复制def benchmark():
    test_cases = [
        ("txt2img", "512x512", 20),
        ("img2img", "768x768", 10),
        ("inpainting", "1024x1024", 5)
    ]
    
    for case in test_cases:
        start = time.time()
        run_workflow(case)
        duration = time.time() - start
        print(f"{case[0]}@{case[1]}: {case[2]/duration:.2f} it/s")

6.2 关键参数优化

1. 调整Spark执行器配置：

python复制spark.conf.set("spark.executor.cores", 4)
spark.conf.set("spark.executor.memoryOverhead", "2g")

2. ComfyUI内存管理：

python复制import gc
gc.collect()
torch.cuda.empty_cache()

7. 故障排查指南

7.1 常见错误代码速查

7.2 日志分析技巧

关键日志位置：

• Spark日志：/opt/spark/logs/
• ComfyUI日志：/opt/ComfyUI/logs/

使用grep快速定位问题：

bash复制grep -i "error" /opt/ComfyUI/logs/*.log
grep -A 5 -B 5 "Exception" /opt/spark/logs/*

8. 高级功能扩展

8.1 自定义节点开发

示例图像后处理节点：

python复制class ImageEnhancer:
    @classmethod
    def INPUT_TYPES(cls):
        return {
            "required": {
                "image": ("IMAGE",),
                "contrast": ("FLOAT", {"default": 1.0}),
            }
        }

    FUNCTION = "enhance"
    CATEGORY = "image/postprocessing"

    def enhance(self, image, contrast):
        # 实现对比度调整逻辑
        return (enhanced_image,)

8.2 分布式工作流设计

将复杂工作流拆分为子任务：

python复制def split_workflow(workflow):
    tasks = []
    for node in workflow["nodes"]:
        if node["type"] == "sampler":
            tasks.append(create_spark_task(node))
    return spark.parallelize(tasks).collect()