Dask 双机集群完整部署指南

作者： /

下面是根据实际验证通过的版本（dask/distributed 2024.6.2 + pandas 2.2.3 + numpy 1.26.4 + pyarrow 21.0.0）整理的完整部署教程。

目录

部署前提

  • 两台机器：10.21.80.38（DSS 主机，UPCDIPPDTKW02）、10.21.80.39（副节点，UPCDIPPDTKW01）
  • 系统：Ubuntu 22.04
  • 用户：pcd_pdtk_op（主组 users，gid=100）
  • 阿里云同 VPC 同安全组，内网带宽 2.7 Gbps
  • DSS 13.3.2 已部署在 38，pyenv 为 Python 3.10（pandas 2.2.3）

关键包版本速查（三端必须一致）

包版本说明
dask2024.6.2核心，支持 pandas 2.x
distributed2024.6.2核心
cloudpickle3.1.2序列化核心，必须显式钉版本
msgpack1.1.2通信协议序列化
lz43.1.3通信压缩
pandas2.2.3数据序列化
numpy1.26.4数据序列化
pyarrow21.0.0列式存储

第 0 阶段：环境检查（两台都做）

# 系统版本
lsb_release -a
​
# 内存大小
free -h
​
# 磁盘空间
df -h /
​
# 互通验证
ping -c 3 10.21.80.38     # 在 39 上执行
ping -c 3 10.21.80.39     # 在 38 上执行

记下两台的实际内存，决定后面 worker 内存限额。

第 1 阶段：Python 环境（两台都做）

DSS pyenv 是 Python 3.10，worker 端使用 Python 3.9（小版本差异兼容，序列化无影响）。

# 检查 Python 3.9 是否可用
python3.9 --version
# 应输出: Python 3.9.x

第 2 阶段：安装 Dask 集群端（两台都做）

# 1. 创建虚拟环境
python3.9 -m venv ~/dask-env
​
# 2. 激活
source ~/dask-env/bin/activate
​
# 3. 升级 pip + 配置阿里云镜像（可选）
pip install --upgrade pip -i https://mirrors.aliyun.com/pypi/simple/
pip config set global.index-url https://mirrors.aliyun.com/pypi/simple/
​
# 4. 安装与 DSS pyenv 对齐的版本
pip install \
    "dask==2024.6.2" \
    "distributed==2024.6.2" \
    "cloudpickle==3.1.2" \
    "msgpack==1.1.2" \
    "lz4==3.1.3" \
    "pandas==2.2.3" \
    "numpy==1.26.4" \
    "pyarrow==21.0.0" \
    "bokeh==2.4.3" \
    "sqlalchemy==2.0.30"
​
# 5. 验证版本（两台输出必须完全一致）
pip list 2>/dev/null | grep -iE "^dask |^distributed|^pandas |^pyarrow|^numpy |^cloudpickle|^msgpack|^lz4 "

预期输出：

cloudpickle     3.1.2
dask            2024.6.2
distributed     2024.6.2
lz4             3.1.3
msgpack         1.1.2
numpy           1.26.4
pandas          2.2.3
pyarrow         21.0.0

⚠️ 重要：

  • 跑 pip check，确认 No broken requirements found.
  • cloudpickle 必须显式指定 3.1.2，不指定的话 pip 可能装其他版本，与 DSS 端不一致会导致 KilledWorker

第 3 阶段：Dask 配置文件（两台都做）

# 创建配置目录
mkdir -p ~/.config/dask
​
# 写入配置
cat > ~/.config/dask/distributed.yaml <<'EOF'
distributed:
  scheduler:
    work-stealing: false      # 关闭 work-stealing，配合机器亲和性调度
​
  worker:
    memory:
      target: 0.75            # 75% 触发主动溢出
      spill: 0.85             # 85% 强制溢出到磁盘
      pause: 0.90             # 90% 暂停接新任务
      terminate: 0.98         # 98% 杀进程
​
  comm:
    compression: lz4          # 跨机传输用 lz4 压缩
    timeouts:
      connect: 30s
      tcp: 30s
EOF
​
# 验证
cat ~/.config/dask/distributed.yaml

第 4 阶段：准备目录（两台都做）

# Spill 目录（如果 /data 有足够空间用 /data，否则用 /tmp）
mkdir -p /tmp/dask-spill
​
# 日志目录
sudo mkdir -p /data/dask
sudo chown pcd_pdtk_op:users /data/dask
sudo chmod 755 /data/dask

第 5 阶段：前台启动测试（先验证能跑）

5.1 机器 38：启动 Scheduler

SSH 到 38，开 tmux：

tmux new -s dask-scheduler
source ~/dask-env/bin/activate
dask scheduler --host 0.0.0.0 --port 8786 --dashboard-address :8787

⚠️ dask 2024.x 推荐用 dask scheduler（空格），dask-scheduler（连字符）也仍然可用

看到如下输出说明成功：

distributed.scheduler - INFO -   Scheduler at:  tcp://10.21.80.38:8786
distributed.scheduler - INFO -   dashboard at:                     :8787

按 Ctrl+B 然后 D 脱离 tmux。

5.2 机器 38：启动 4 个主力 Worker

新开终端：

tmux new -s dask-worker
source ~/dask-env/bin/activate
​
dask worker tcp://10.21.80.38:8786 \
    --nworkers 4 \
    --nthreads 4 \
    --memory-limit 60GB \
    --local-directory /tmp/dask-spill \
    --resources "primary=1 machine_38=1" \
    --name primary

⚠️ 关键点：

  • dask 2024.x 用 dask worker（空格）或 dask-worker（连字符），两者都可用
  • 必须用 --nworkers（--nprocs 在 2024.x 中已移除）
  • --name primary（自动追加 -0、-1、-2、-3）
  • --resources "primary=1 machine_38=1"：machine_38 标签用于 Job 级别机器亲和性调度

看到 4 行 Registered to: tcp://10.21.80.38:8786 说明 OK。

5.3 机器 39：启动 Secondary Worker

SSH 到 39：

tmux new -s dask-worker
source ~/dask-env/bin/activate
​
dask worker tcp://10.21.80.38:8786 \
    --nworkers 1 \
    --nthreads 4 \
    --memory-limit 50GB \
    --local-directory /tmp/dask-spill \
    --resources "secondary=1 machine_39=1" \
    --name secondary-39

5.4 验证集群

回到 38 任意终端：

source ~/dask-env/bin/activate
python3 -c "
from dask.distributed import Client
c = Client('tcp://10.21.80.38:8786')
info = c.scheduler_info()
print(f'Workers: {len(info[\"workers\"])}/5')
for w in info['workers'].values():
    res = w.get('resources', {})
    mem = w['memory_limit']/1e9
    print(f'  {w[\"name\"]}: nthreads={w[\"nthreads\"]}, mem={mem:.0f}GB, resources={res}')
"

预期：

Workers: 5/5
  primary-0: nthreads=4, mem=60GB, resources={'primary': 1.0, 'machine_38': 1.0}
  primary-1: nthreads=4, mem=60GB, resources={'primary': 1.0, 'machine_38': 1.0}
  primary-2: nthreads=4, mem=60GB, resources={'primary': 1.0, 'machine_38': 1.0}
  primary-3: nthreads=4, mem=60GB, resources={'primary': 1.0, 'machine_38': 1.0}
  secondary-39: nthreads=4, mem=XXGB, resources={'secondary': 1.0, 'machine_39': 1.0}

5.5 简单计算测试

import dask.array as da
from dask.distributed import Client
from dask import annotate
​
c = Client("tcp://10.21.80.38:8786")
with annotate(resources={'primary': 1}):
    x = da.random.random((50000, 25000), chunks=(5000, 5000))
    result = x.mean().compute()
print(f"Mean: {result}")
print("✓ 集群工作正常")

跑通后进入 tmux 各窗口按 Ctrl+C 关掉前台进程（或 tmux kill-server），准备做 systemd 部署。

第 6 阶段：systemd 服务部署

6.1 机器 38：Scheduler Service

sudo tee /etc/systemd/system/dask-scheduler.service > /dev/null <<'EOF'
[Unit]
Description=Dask Scheduler
After=network.target
Wants=network-online.target
​
[Service]
Type=simple
User=pcd_pdtk_op
Group=users
WorkingDirectory=/home/pcd_pdtk_op
Environment="PATH=/home/pcd_pdtk_op/dask-env/bin:/usr/local/bin:/usr/bin:/bin"
ExecStart=/home/pcd_pdtk_op/dask-env/bin/dask-scheduler \
    --host 0.0.0.0 \
    --port 8786 \
    --dashboard-address :8787
Restart=always
RestartSec=10
StandardOutput=append:/data/dask/scheduler.log
StandardError=append:/data/dask/scheduler.err
​
[Install]
WantedBy=multi-user.target
EOF

⚠️ 关键点：

  • Group=users（不是 Group=pcd_pdtk_op）
  • systemd ExecStart 建议用 dask-scheduler（连字符），直接指向 bin 下的可执行文件
  • 日志路径 /data/dask/

6.2 机器 38：Worker Service

sudo tee /etc/systemd/system/dask-worker.service > /dev/null <<'EOF'
[Unit]
Description=Dask Worker (Primary on 38)
After=network.target dask-scheduler.service
Requires=dask-scheduler.service
​
[Service]
Type=simple
User=pcd_pdtk_op
Group=users
WorkingDirectory=/home/pcd_pdtk_op
Environment="PATH=/home/pcd_pdtk_op/dask-env/bin:/usr/local/bin:/usr/bin:/bin"
ExecStart=/home/pcd_pdtk_op/dask-env/bin/dask-worker tcp://10.21.80.38:8786 \
    --nworkers 4 \
    --nthreads 4 \
    --memory-limit 60GB \
    --local-directory /tmp/dask-spill \
    --resources "primary=1 machine_38=1" \
    --name primary
Restart=always
RestartSec=15
StandardOutput=append:/data/dask/worker.log
StandardError=append:/data/dask/worker.err
LimitNOFILE=65535
​
[Install]
WantedBy=multi-user.target
EOF

⚠️ dask 2024.x 必须用 --nworkers，--nprocs 已被移除，会导致 exit code 2。

6.3 机器 39：Worker Service

SSH 到 39：

sudo tee /etc/systemd/system/dask-worker.service > /dev/null <<'EOF'
[Unit]
Description=Dask Worker (Secondary on 39)
After=network.target
Wants=network-online.target
​
[Service]
Type=simple
User=pcd_pdtk_op
Group=users
WorkingDirectory=/home/pcd_pdtk_op
Environment="PATH=/home/pcd_pdtk_op/dask-env/bin:/usr/local/bin:/usr/bin:/bin"
ExecStart=/home/pcd_pdtk_op/dask-env/bin/dask-worker tcp://10.21.80.38:8786 \
    --nworkers 1 \
    --nthreads 4 \
    --memory-limit 50GB \
    --local-directory /tmp/dask-spill \
    --resources "secondary=1 machine_39=1" \
    --name secondary-39
Restart=always
RestartSec=15
StandardOutput=append:/data/dask/worker.log
StandardError=append:/data/dask/worker.err
LimitNOFILE=65535
​
[Install]
WantedBy=multi-user.target
EOF

6.4 启动服务

机器 38：

sudo systemctl daemon-reload
sudo systemctl enable dask-scheduler dask-worker
sudo systemctl start dask-scheduler
sleep 5
sudo systemctl start dask-worker
​
# 查看状态
sudo systemctl status dask-scheduler --no-pager
sudo systemctl status dask-worker --no-pager

机器 39：

sudo systemctl daemon-reload
sudo systemctl enable dask-worker
sudo systemctl start dask-worker
​
sudo systemctl status dask-worker --no-pager

均应看到 Active: active (running)。

6.5 验证服务化集群

source ~/dask-env/bin/activate
python3 -c "
from dask.distributed import Client
c = Client('tcp://10.21.80.38:8786')
info = c.scheduler_info()
print(f'Workers: {len(info[\"workers\"])}/5')
for w in info['workers'].values():
    print(f'  {w[\"name\"]}: {w[\"memory_limit\"]/1e9:.0f}GB, {w.get(\"resources\")}')
"

第 7 阶段：DSS Code Env 集成

7.1 方式 A：使用 DSS 内置 pyenv（推荐）

如果 DSS 的 pyenv（/data/DATA_DIR/pyenv）已包含 pandas 2.2.3、numpy 1.26.4，只需确认 dask 相关包已安装：

/data/DATA_DIR/pyenv/bin/pip install \
    "dask==2024.6.2" \
    "distributed==2024.6.2" \
    "cloudpickle==3.1.2" \
    "msgpack==1.1.2" \
    "lz4==3.1.3"

7.2 方式 B：通过 DSS Code Env 管理界面

Administration → Code envs → 对应 code env → Packages to install

在已有 requirements 末尾追加：

dask==2024.6.2
distributed==2024.6.2
cloudpickle==3.1.2
msgpack==1.1.2
lz4==3.1.3

⚠️ 必须显式写 cloudpickle、msgpack、lz4：

  • cloudpickle 不钉版本的话 pip 可能解析到其他版本，与 worker 端不一致会导致 KilledWorker
  • msgpack 和 lz4 同理，钉死确保三端一致

⚠️ 不要写：

  • pandas / pyarrow / numpy（已由 DSS 环境锁定，不要覆盖）
  • bokeh（client 端不需要）

点 Save and update，等 2-5 分钟。

7.3 处理 GitHub 依赖拉取失败（如遇到）

若 code env 含 GitHub 私有库依赖（如 lvmh-group/atom-ds-core-library），可能因 GnuTLS 报错。处理方案：

# 方案 A：重试几次
# 方案 B：给 git 配代理（DSS 运行用户身份执行）
git config --global http.proxy http://127.0.0.1:7890
git config --global https.proxy http://127.0.0.1:7890
​
# 方案 C：用 ghproxy 镜像
git config --global url."https://ghproxy.com/https://github.com/".insteadOf "https://github.com/"

7.4 测试 DSS 到集群的连接

DSS 里新建 Python recipe：

import pandas, dask, distributed, cloudpickle, msgpack
print(f"pandas:      {pandas.__version__}        (期望 2.2.3)")
print(f"numpy:       {__import__('numpy').__version__}       (期望 1.26.4)")
print(f"dask:        {dask.__version__}          (期望 2024.6.2)")
print(f"distributed: {distributed.__version__}   (期望 2024.6.2)")
print(f"cloudpickle: {cloudpickle.__version__}   (期望 3.1.2)")
print(f"msgpack:     {msgpack.version}           (期望 (1, 1, 2))")
​
from dask.distributed import Client
c = Client("tcp://10.21.80.38:8786")
print(c)
​
info = c.scheduler_info()
print(f"\nWorkers: {len(info['workers'])}/5")
for w in info['workers'].values():
    print(f"  {w['name']}: {w['memory_limit']/1e9:.0f}GB, {w.get('resources')}")
​
c.close()

第 8 阶段：日常运维

启动 / 停止 / 重启

# 38 上
sudo systemctl restart dask-scheduler
sudo systemctl restart dask-worker
sudo systemctl stop dask-worker
​
# 39 上
sudo systemctl restart dask-worker

查看日志

# systemd 日志
sudo journalctl -u dask-scheduler -f
sudo journalctl -u dask-worker -f
​
# 应用日志文件
tail -f /data/dask/scheduler.log
tail -f /data/dask/worker.log

Dashboard 监控

浏览器访问 http://10.21.80.38:8787（需阿里云安全组放通 8787 入方向）

本地无法直接访问可用 SSH 端口转发：

ssh -L 8787:10.21.80.38:8787 pcd_pdtk_op@10.21.80.38
# 浏览器打开 http://localhost:8787

完整集群重启

# 在 39 上停 worker
ssh pcd_pdtk_op@10.21.80.39 "sudo systemctl stop dask-worker"
​
# 在 38 上重启
sudo systemctl restart dask-scheduler
sleep 5
sudo systemctl restart dask-worker
​
# 在 39 上启 worker
ssh pcd_pdtk_op@10.21.80.39 "sudo systemctl start dask-worker"

第 9 阶段：DSS Recipe 模板（Job 级别机器亲和性调度）

"""
Dask on DSS — Job 级别机器亲和性调度
====================================
调度逻辑：
  - 每个 Job 自动选最空闲的机器
  - scatter + annotate 双重钉机器，数据零跨机传输
  - 2 个 Job 并行时各占一台
"""
import dataiku
import dask
import dask.dataframe as dd
import pandas as pd
import numpy as np
from dask.distributed import Client, wait
from dask import delayed
​
# ============ 配置 ============
SCHEDULER_ADDR = "tcp://10.21.80.38:8786"
MACHINE_TAGS   = ["machine_38", "machine_39"]
OUTPUT_DATASET = "your_output_dataset"  # ← 改成你 recipe 的 output dataset 名
NUM_TEST_ROWS  = 2_000_000              # 测试数据行数，正式使用时替换为 DSS 数据读取
CHUNK_SIZE     = 500_000
​
# ============ 1. 连接 ============
client = Client(SCHEDULER_ADDR)
print(client)
​
# ============ 2. 版本校验 ============
def get_worker_versions():
    import sys, pandas, numpy, dask, distributed, cloudpickle
    return {
        "python": f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}",
        "pandas": pandas.__version__,
        "numpy": numpy.__version__,
        "dask": dask.__version__,
        "distributed": distributed.__version__,
        "cloudpickle": cloudpickle.__version__,
    }
​
import sys
client_ver = {
    "python": f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}",
    "pandas": pd.__version__, "numpy": np.__version__,
    "dask": dask.__version__, "distributed": __import__('distributed').__version__,
    "cloudpickle": __import__('cloudpickle').__version__,
}
​
worker_results = client.run(get_worker_versions)
if not worker_results:
    raise RuntimeError("没有 worker 连接到集群，请检查 worker 服务状态")
worker_ver = list(worker_results.values())[0]
​
print(f"\n{'Package':<15} {'Client':<18} {'Worker':<18} {'Match'}")
print("-" * 60)
for pkg in client_ver:
    cv, wv = client_ver[pkg], worker_ver.get(pkg, "?")
    print(f"{pkg:<15} {cv:<18} {wv:<18} {'✓' if cv == wv else '✗'}")
print()
​
# ============ 3. 选机器 ============
def pick_machine(client, machine_tags):
    workers = client.scheduler_info()["workers"]
    machine_stats = {}
    for tag in machine_tags:
        addrs, free_mem = [], 0
        for addr, info in workers.items():
            if info.get("resources", {}).get(tag, 0) > 0:
                addrs.append(addr)
                ml = info.get("memory_limit", 0)
                m = info.get("metrics", {})
                free_mem += max(0, ml - m.get("managed_bytes", 0) - m.get("unmanaged_old_bytes", 0))
        if addrs:
            machine_stats[tag] = {"addrs": addrs, "free_mem": free_mem}
​
    if not machine_stats:
        raise RuntimeError(f"没有带 {machine_tags} 标签的 worker")
​
    for tag, s in machine_stats.items():
        print(f"  {tag}: {len(s['addrs'])} workers, 可用 {s['free_mem']/1e9:.1f} GB")
​
    chosen_tag = max(machine_stats, key=lambda t: machine_stats[t]["free_mem"])
    chosen = machine_stats[chosen_tag]
    print(f"✓ 选定: {chosen_tag} ({len(chosen['addrs'])} workers, {chosen['free_mem']/1e9:.1f} GB)\n")
    return chosen_tag, chosen["addrs"]
​
machine_tag, target_workers = pick_machine(client, MACHINE_TAGS)
​
# ============ 4A. 生成测试数据（验证用，正式使用时替换为 4B） ============
print(f"生成 {NUM_TEST_ROWS:,} 行测试数据...")
rng = np.random.RandomState(42)
​
futures = []
rows_done = 0
chunk_id = 0
while rows_done < NUM_TEST_ROWS:
    n = min(CHUNK_SIZE, NUM_TEST_ROWS - rows_done)
    chunk = pd.DataFrame({
        "store_id": rng.randint(1, 101, size=n),
        "amount":   rng.uniform(10, 500, size=n).round(2),
        "qty":      rng.randint(1, 20, size=n),
        "date":     pd.Timestamp("2025-01-01") + pd.to_timedelta(rng.randint(0, 365, size=n), unit="D"),
    })
    future = client.scatter(chunk, workers=target_workers, broadcast=False)
    futures.append(future)
    rows_done += n
    chunk_id += 1
    del chunk
​
# ============ 4B. 从 DSS Dataset 读取（正式使用时取消注释，删除 4A） ============
# INPUT_DATASET = "your_input_dataset"  # ← 改成你 recipe 的 input dataset 名
# input_ds = dataiku.Dataset(INPUT_DATASET)
# futures = []
# chunk_id = 0
# for chunk in input_ds.iter_dataframes(chunksize=CHUNK_SIZE):
#     future = client.scatter(chunk, workers=target_workers, broadcast=False)
#     futures.append(future)
#     chunk_id += 1
#     del chunk
​
wait(futures)
print(f"✓ scatter 完成: {chunk_id} chunks → {len(target_workers)} workers")
​
# 验证数据分布
workers_info = client.scheduler_info()["workers"]
w2m = {}
for tag in MACHINE_TAGS:
    for addr, info in workers_info.items():
        if info.get("resources", {}).get(tag, 0) > 0:
            w2m[addr] = tag
​
tag_counts = {t: 0 for t in MACHINE_TAGS}
for f in futures:
    for key, addrs in client.who_has(f).items():
        for a in addrs:
            t = w2m.get(a, "unknown")
            if t in tag_counts:
                tag_counts[t] += 1
​
for t, c in tag_counts.items():
    print(f"  {t}: {c} partitions")
​
# 构建 dask DataFrame
meta = futures[0].result().iloc[:0]
ddf = dd.from_delayed([delayed(lambda f: f)(f) for f in futures], meta=meta)
print(f"✓ dask DataFrame: {ddf.npartitions} partitions\n")
​
# ============ 5. 业务计算（钉在目标机器） ============
print("开始计算...")
with dask.annotate(resources={machine_tag: 1}):
    result = ddf.groupby("store_id").agg({
        "amount": "sum",
        "qty":    "sum",
    }).reset_index()
​
# ============ 6. 写出到 DSS ============
# 方式 A：结果较小（< 1GB），直接 compute 回本地写入
with dask.annotate(resources={machine_tag: 1}):
    result_pd = result.compute()
​
print(f"✓ 计算完成: {len(result_pd)} 行, "
      f"{result_pd.memory_usage(deep=True).sum()/1e6:.1f} MB")
print(result_pd.head(10))
​
output_ds = dataiku.Dataset(OUTPUT_DATASET)
output_ds.write_with_schema(result_pd)
print(f"\n✓ 已写入 {OUTPUT_DATASET}")
​
# 方式 B：结果较大（> 1GB），分批写，避免 DSS recipe 进程 OOM
# output_ds = dataiku.Dataset(OUTPUT_DATASET)
# output_ds.write_schema_from_dataframe(result.head(100))
# with output_ds.get_writer() as writer:
#     for part in result.to_delayed():
#         with dask.annotate(resources={machine_tag: 1}):
#             part_pd = part.compute()
#         writer.write_dataframe(part_pd)
#         del part_pd
​
client.close()
print("✓ Done")

第 10 阶段：故障排查速查表

症状检查/处理
systemd 启动 status=216/GROUPGroup 不存在，改 Group=users
Worker exit code=2--nprocs 在 2024.x 中已移除，改成 --nworkers
Worker 连不上 schedulersudo systemctl status dask-scheduler + 阿里云安全组 8786 端口
KilledWorker / 版本协议错误pip list 对比 cloudpickle、msgpack、lz4 三端必须完全一致
StringMethods AttributeErrordask 版本过旧不兼容 pandas 2.x，升级到 dask 2024.1.0+
IndexError: list index out of range (版本校验时)没有 worker 连上来，先检查 scheduler 和 worker 服务状态
Worker 频繁重启sudo journalctl -u dask-worker -n 100 看错误
annotate 后 task 挂起不执行worker --resources 缺少 machine tag，检查是否有 machine_38=1 / machine_39=1
任务在 39 上跑了38 内存满了或 partition 调度不平衡，是正常溢出行为
任务全在 38 上跑，39 不动正常，39 是溢出补充用的
Dashboard 打不开阿里云安全组放通 8787 入方向
compute() 时 DSS OOM结果太大，改用分批写（方式 B）或 to_parquet 写盘
Dataset xxx cannot be useddataset 名没在 recipe 的 Input/Output 面板里声明，或名字拼错了
插件 GitHub 依赖拉取失败配 git proxy 或用 ghproxy 镜像

第 11 阶段：阿里云安全组

两台 ECS 加入同一安全组，组内规则放通：

协议端口授权对象
TCP8786/8787同安全组
TCP1024-65535同安全组（worker 随机端口）
ICMP全部同安全组

最终验证清单

  • 两台 pip list 关键包版本完全一致（dask / distributed / cloudpickle / msgpack / lz4 / pandas / numpy / pyarrow）
  • ~/.config/dask/distributed.yaml 两台存在且一致
  • 38 上 dask-scheduler.service 和 dask-worker.service running
  • 39 上 dask-worker.service running
  • Worker 使用 --nworkers（不是 --nprocs）
  • Worker resources 包含 machine tag（machine_38=1 / machine_39=1）
  • Python 客户端能看到 5 个 worker
  • Dashboard http://10.21.80.38:8787 能访问
  • DSS code env 装好 dask / distributed / cloudpickle / msgpack / lz4
  • DSS recipe 里能 Client("tcp://10.21.80.38:8786") 成功
  • 重启服务后能自动注册到集群

关键注意事项汇总

⚠️ dask 2024.6.2 相比 2022.5.0 的变化：

  • --nprocs 已移除，必须用 --nworkers
  • dask scheduler / dask worker（空格命令）是推荐写法，连字符写法仍兼容
  • 原生支持 pandas 2.x，不再有 StringMethods 兼容性问题
  • 支持 P2P shuffle，大表 groupby/join 性能更好

⚠️ DSS Code Env 的约束：

  • cloudpickle==3.1.2、msgpack==1.1.2、lz4==3.1.3 必须显式钉版本
  • pandas / numpy / pyarrow 由 DSS 环境管理，不要在 dask 配置中覆盖
  • 数据读取走 DSS API + scatter，不走 dd.read_sql_*

⚠️ 机器亲和性调度：

  • Worker 必须带 machine_38=1 / machine_39=1 资源标签
  • Recipe 代码通过 MACHINE_TAGS 引用这些标签
  • scatter(workers=target_workers) 钉数据位置
  • dask.annotate(resources={machine_tag: 1}) 钉计算位置
  • 两个 Job 并行时自动按可用内存分配到不同机器

⚠️ 优化建议：

  • SQL 端预聚合优先于 Dask shuffle
  • partition 大小控制在 100-500 MB
  • 结果大于 1GB 用分批写（方式 B）而不是 compute() 收回 client

第 12 阶段：生产内存保护与监控加固

12.1 故障背景

2026-08-12，10.21.80.38 发生 Linux 全局 OOM：

  • 多个 UID 1005 的 Python/Dask 进程占用大量内存；
  • 内核多次执行 Out of memory: Killed process；
  • Dask Worker Event Loop 长时间无响应；
  • Dataiku 端口 11000 健康检查失败；
  • 重启服务器后 Dataiku 恢复。

原配置：

38：4 workers × 60GB
39：1 worker × 50GB

Dask 的 --memory-limit 按每个 Worker 计算，38 节点理论上可使用约 240GB，与 Dataiku JVM、系统及其他进程发生内存争抢。

12.2 统一后的生产配置

两台服务器统一使用：

Dask 版本：2024.6.2
Workers：4
Threads per worker：4
Memory limit per worker：40GB
每台服务器 Dask 总目标内存：4 × 40GB = 160GB

Dask 日志中每个 Worker 显示约：

37.25GiB

节点配置：

节点地址Worker 名称资源标签
主节点10.21.80.38primary-0 ~ primary-3primary=1 machine_38=1
副节点10.21.80.39secondary-39-0 ~ secondary-39-3secondary=1 machine_39=1

集群总状态：

8 workers
32 threads
每个 Worker 37.25GiB

12.3 Spill 目录

两台服务器均使用数据盘：

sudo install -d -o pcd_pdtk_op -g users -m 0750 /data/dask-spill

Worker 启动参数：

--local-directory /data/dask-spill

不再使用：

/tmp/dask-spill

12.4 systemd 内存保护

两台服务器的 dask-worker.service 均增加：

MemoryAccounting=yes
MemoryHigh=160G
MemoryMax=175G

说明：

  • MemoryHigh=160G：达到后触发内存回收压力；
  • MemoryMax=175G：Dask 服务硬限制；
  • 限制作用于整个 Dask Worker 服务；
  • 防止异常任务拖垮整台服务器。

查看配置：

sudo systemctl show dask-worker.service \
  -p MemoryAccounting -p MemoryHigh -p MemoryMax

预期：

MemoryAccounting=yes
MemoryHigh=171798691840
MemoryMax=187904819200

12.5 38 节点 Worker 配置

ExecStart=/home/pcd_pdtk_op/dask-env/bin/dask-worker tcp://10.21.80.38:8786 --nworkers 4 --nthreads 4 --memory-limit 40GB --local-directory /data/dask-spill --resources "primary=1 machine_38=1" --name primary

12.6 39 节点 Worker 配置

ExecStart=/home/pcd_pdtk_op/dask-env/bin/dask-worker tcp://10.21.80.38:8786 --nworkers 4 --nthreads 4 --memory-limit 40GB --local-directory /data/dask-spill --resources "secondary=1 machine_39=1" --name secondary-39

12.7 Dask 内存阈值

两台服务器的：

/home/pcd_pdtk_op/.config/dask/distributed.yaml

统一配置为：

distributed:
  scheduler:
    work-stealing: false
​
  worker:
    memory:
      target: 0.75
      spill: 0.85
      pause: 0.90
      terminate: 0.98
​
  comm:
    compression: lz4
    timeouts:
      connect: 30s
      tcp: 30s

12.8 服务重载与验证

sudo systemctl daemon-reload
sudo systemctl restart dask-worker.service

在 38 上验证集群：

/home/pcd_pdtk_op/dask-env/bin/python - <<'PY'
from dask.distributed import Client
​
c = Client("tcp://10.21.80.38:8786", timeout="10s")
info = c.scheduler_info()
print("workers:", len(info["workers"]))
​
for address, w in sorted(info["workers"].items(), key=lambda x: x[1]["name"]):
    print(
        f'{w["name"]}: '
        f'threads={w["nthreads"]}, '
        f'memory={w["memory_limit"]/2**30:.2f}GiB, '
        f'resources={w.get("resources", {})}'
    )
​
c.close()
PY

预期：

workers: 8
每个 Worker: threads=4
每个 Worker: memory=37.25GiB

验证 Dataiku：

curl -sS -o /dev/null -w '11000 HTTP %{http_code}\n' \
  http://10.21.80.38:11000/

预期：

11000 HTTP 200

验证 spill 路径：

sudo grep -E 'Worker name:|Memory:|Local Directory:' \
  /data/dask/worker.err | tail -24

12.9 主机监控脚本

监控脚本：

/data/check.sh

监控日志：

/data/daily_check.log

监控内容：

  • CPU 使用率；
  • 系统负载；
  • 内存使用率；
  • Swap 使用率；
  • 内存占用最高进程；
  • CPU 占用最高进程；
  • Python 进程；
  • 最近 10 分钟 OOM 日志。

Cron 配置：

*/5 * * * * /data/check.sh

脚本使用锁文件防止重复执行：

/tmp/daily_check.lock

确认任务：

sudo crontab -u pcd_pdtk_op -l

12.10 监控日志轮转

配置文件：

/etc/logrotate.d/daily-check

内容：

/data/daily_check.log {
    daily
    rotate 14
    maxsize 100M
    compress
    delaycompress
    missingok
    notifempty
    copytruncate
}

说明：

  • 每天检查一次；
  • 日志超过 100MB 时提前轮转；
  • 保留 14 份；
  • 历史日志压缩；
  • 不需要重启监控脚本。

验证：

sudo logrotate -d /etc/logrotate.d/daily-check

12.11 日常检查命令

free -h
sudo journalctl -k --since '1 hour ago' --no-pager | \
grep -Ei 'oom|out of memory|killed process'
ps -eo pid,ppid,user,%cpu,%mem,rss,etime,cmd --sort=-rss | head -30
sudo systemctl status dask-worker.service --no-pager
sudo grep -Ei 'signal 9|signal 15|memory|spill|pause|terminate|unresponsive' \
  /data/dask/worker.err | tail -100

12.12 回滚方案

如需恢复原 Worker 配置：

sudo rm -f /etc/systemd/system/dask-worker.service.d/override.conf
sudo systemctl daemon-reload
sudo systemctl restart dask-worker.service

如需恢复原日志查询窗口：

sudo sed -i 's/--since "10 minutes ago"/--since "1 hour ago"/' /data/check.sh

12.13 当前验证结论

当前双机 Dask 集群已验证：

38 和 39 共 8 个 Worker
每个 Worker 4 个线程
每个 Worker 37.25GiB
spill 使用 /data/dask-spill
Dask 服务具备 MemoryHigh/MemoryMax 保护
Dataiku 11000 返回 HTTP 200

该配置用于避免 Dask 异常任务再次触发主机级 OOM。

滚动至顶部