文档中心

Appearance

✨费用中心控制台

创建数据处理任务 ​

数据处理任务适合在正式训练前执行 CPU 侧工作，例如数据清洗、格式转换、样本索引生成、离线评估或小规模预处理。该类型不使用 GPU，且优先保障 GPU 相关任务不受影响。

确认任务适合放到数据处理任务中 ​

适合使用数据处理任务的工作：

  • 解压、去重、切分或转换数据集。
  • 生成训练样本索引、manifest 或 metadata。
  • 校验样本字段、图片路径、音视频时长或 tokenizer 输出。
  • 把预处理结果写入共享高性能存储，供后续训练任务读取。

不适合使用数据处理任务的工作：

  • 需要 GPU 的训练、微调或推理。
  • 需要多 Worker GPU 通信的分布式训练。
  • 需要长时间独占高性能 GPU 节点的任务。

创建数据处理任务 ​

本节说明如何在不占用 GPU 的情况下运行数据处理脚本，并把处理结果写入共享高性能存储。以下示例把 Alpaca 原始 Parquet 数据转换为训练任务可直接读取的 chat messages JSONL。

创建任务

以下步骤为完整创建流程。

Step 0 选择数据处理任务 ​

在 任务类型 中选择 数据处理任务。该类型面向 CPU 数据处理，不配置分布式训练框架，也不使用 GPU。

Step 1 选择资源规格 ​

选择能满足脚本 CPU 和内存需求的规格。数据处理任务通常不需要多 Worker；如果脚本需要并发处理，请优先在脚本内部控制并发数，避免对共享存储造成过高压力。

Step 2 准备转换脚本 ​

处理逻辑建议写在单独脚本中，并放到共享高性能存储或构建进镜像。以下示例会读取 Alpaca 的 Parquet 文件，校验 instruction 和 output 字段，将样本转换为 chat messages JSONL，并写出 train.jsonl、valid.jsonl 和 summary.json。

开始前，先将 Stanford Alpaca（tatsu-lab/alpaca）上传到共享高性能存储。以下路径仅为示例，创建任务时请替换为自己的共享存储路径：

language-text
/mnt/zhaoyinghao/datasets/alpaca/raw/alpaca.20260604-161913/

先在开发机或 AICoder 中准备转换脚本：

language-text
/mnt/zhaoyinghao/projects/alpaca-processing/scripts/convert_alpaca_to_chat_jsonl.py

脚本内容如下：

language-python
#!/usr/bin/env python3
import argparse
import json
from pathlib import Path

import pyarrow.parquet as pq


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--raw-dir", required=True)
    parser.add_argument("--output-dir", required=True)
    parser.add_argument("--valid-every", type=int, default=50)
    args = parser.parse_args()

    if args.valid_every < 2:
        raise SystemExit("--valid-every must be greater than 1")

    raw_dir = Path(args.raw_dir)
    output_dir = Path(args.output_dir)
    data_dir = raw_dir / "data"

    if not data_dir.is_dir():
        raise SystemExit(f"Data directory not found: {data_dir}")

    output_dir.mkdir(parents=True, exist_ok=False)

    parquet_files = sorted(data_dir.glob("*.parquet"))
    if not parquet_files:
        raise SystemExit(f"No parquet files found in {data_dir}")

    train_path = output_dir / "train.jsonl"
    valid_path = output_dir / "valid.jsonl"
    summary_path = output_dir / "summary.json"

    total = 0
    train_count = 0
    valid_count = 0
    skipped = 0

    with train_path.open("w", encoding="utf-8") as train_out, valid_path.open("w", encoding="utf-8") as valid_out:
        for parquet_file in parquet_files:
            pf = pq.ParquetFile(parquet_file)
            missing = {"instruction", "output"} - set(pf.schema_arrow.names)
            if missing:
                raise SystemExit(f"{parquet_file} missing required columns: {sorted(missing)}")

            for batch in pf.iter_batches(batch_size=5000):
                for row in batch.to_pylist():
                    instruction = (row.get("instruction") or "").strip()
                    input_text = (row.get("input") or "").strip()
                    output = (row.get("output") or "").strip()

                    if not instruction or not output:
                        skipped += 1
                        continue

                    user_content = instruction if not input_text else f"{instruction}\n\n{input_text}"
                    record = {
                        "messages": [
                            {"role": "user", "content": user_content},
                            {"role": "assistant", "content": output},
                        ]
                    }

                    total += 1
                    line = json.dumps(record, ensure_ascii=False) + "\n"
                    if total % args.valid_every == 0:
                        valid_out.write(line)
                        valid_count += 1
                    else:
                        train_out.write(line)
                        train_count += 1

    if total == 0:
        raise SystemExit("No valid records converted")

    summary = {
        "raw_dir": str(raw_dir),
        "parquet_files": [str(p) for p in parquet_files],
        "format": "chat messages JSONL",
        "valid_every": args.valid_every,
        "total_converted": total,
        "train_count": train_count,
        "valid_count": valid_count,
        "skipped": skipped,
        "train_path": str(train_path),
        "valid_path": str(valid_path),
    }
    summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
    print(json.dumps(summary, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

Step 3 填写启动命令 ​

启动命令 只负责设置路径、准备运行环境、调用脚本和发布结果。复制以下示例时，先确认 RAW_DIR、OUTPUT_BASE 和 SCRIPT 指向本次任务使用的共享存储路径。

如果使用通用基础镜像，先确认镜像已经包含启动命令依赖的系统包、Python 包和虚拟环境工具。更稳定的做法是提前准备包含转换依赖的镜像，避免任务运行时临时安装依赖。

language-bash
set -euo pipefail

RAW_DIR="/mnt/zhaoyinghao/datasets/alpaca/raw/alpaca.20260604-161913"
OUTPUT_BASE="/mnt/zhaoyinghao/datasets/alpaca/processed"
SCRIPT="/mnt/zhaoyinghao/projects/alpaca-processing/scripts/convert_alpaca_to_chat_jsonl.py"
RUN_ID="${RUN_ID:-$(date +%Y%m%d-%H%M%S)}"

WORK_DIR="${OUTPUT_BASE}/alpaca-chat.${RUN_ID}.tmp"
FINAL_DIR="${OUTPUT_BASE}/alpaca-chat.${RUN_ID}"
LOG_DIR="${OUTPUT_BASE}/logs"

if [ -e "$WORK_DIR" ] || [ -e "$FINAL_DIR" ]; then
  echo "Output path already exists: $WORK_DIR or $FINAL_DIR"
  exit 1
fi

mkdir -p "$OUTPUT_BASE" "$LOG_DIR"

python3 -m venv /tmp/alpaca-processing-venv
/tmp/alpaca-processing-venv/bin/python -m pip install --upgrade pip
/tmp/alpaca-processing-venv/bin/python -m pip install --no-cache-dir -U pyarrow

/tmp/alpaca-processing-venv/bin/python "$SCRIPT" \
  --raw-dir "$RAW_DIR" \
  --output-dir "$WORK_DIR" \
  2>&1 | tee "$LOG_DIR/alpaca-chat.${RUN_ID}.log"

mv "$WORK_DIR" "$FINAL_DIR"

echo "Converted dataset written to: $FINAL_DIR"
wc -l "$FINAL_DIR/train.jsonl" "$FINAL_DIR/valid.jsonl"
cat "$FINAL_DIR/summary.json"

使用该 启动命令 时，注意以下做法：

  • 保持 RAW_DIR 只读使用，不在原始数据目录中写入中间文件或处理结果。
  • 每次运行使用新的 RUN_ID 和输出目录；手动重跑任务时，不会覆盖上一次已经完成的结果。
  • 处理中间结果先写入 .tmp 目录，脚本成功后再发布为最终目录。看到 .tmp 目录时，应优先按未完成结果处理。
  • 保留 set -euo pipefail，让转换脚本失败时，数据处理任务也显示失败；否则日志里已经报错，任务状态仍可能显示成功。详见编写可靠启动命令。
  • 示例命令会在任务运行时创建虚拟环境并安装 pyarrow。如果镜像缺少虚拟环境工具、pip 或必要系统包，任务会在转换脚本开始前失败。反复使用该转换流程时，建议准备包含 pyarrow 的镜像，并删除 启动命令 中的 pip install 部分。

Step 4 挂载输入和输出存储 ​

如果输入数据、转换脚本和输出结果都在共享高性能存储中，创建任务时挂载同一个存储卷，并保持与后续训练任务相同的挂载路径。

公共数据目录通常只读，只适合作为输入；处理结果应写入共享高性能存储。

Step 5 创建后检查结果 ​

任务完成后，使用 AICoder 或开发机挂载同一存储卷，检查输出文件数量、大小和样本格式。确认数据处理结果可读后，再继续创建训练任务。