LLM学習データの準備：FineWebの100GBスライスをS3に置いてみた

uv venv --seed --python 3.12
uv pip install 'skypilot[aws]'

uv run sky check aws

🎉 Enabled infra 🎉
  AWS [compute, storage]

test/
├── .sky.yaml                 # SkyPilot project config (auth method)
├── fineweb.yaml              # SkyPilot task (instance + run steps)
├── stream_fineweb_slice.py   # hf:// -> s3:// streaming
├── build_manifest.py         # build the manifest after upload
└── infra/
    └── main.tf               # S3 bucket (Terraform)

# infra/main.tf
terraform {
  required_providers {
    aws    = { source = "hashicorp/aws", version = "~> 5.0" }
    random = { source = "hashicorp/random", version = "~> 3.0" }
  }
}

provider "aws" {
  region = var.region
}

variable "region" {
  default = "ap-northeast-1"
}

variable "bucket_prefix" {
  default = "fineweb-mirror"
}

# S3 bucket names are globally unique; a random suffix avoids collisions
resource "random_id" "suffix" {
  byte_length = 4 # -> 8 hex chars, e.g. "a3f9c1d0"
}

resource "aws_s3_bucket" "fineweb" {
  bucket = "${var.bucket_prefix}-${random_id.suffix.hex}"

  # allow destroy even when the bucket still holds ~100GB (test env)
  force_destroy = true
}

output "bucket_name" {
  value = aws_s3_bucket.fineweb.bucket
}

cd test/infra
terraform init
terraform apply                # create the bucket
terraform output bucket_name   # pass this name to sky launch

# stream_fineweb_slice.py
import argparse
import concurrent.futures
import os
import shutil
import sys

import fsspec
from huggingface_hub import HfFileSystem

HF_REPO = "datasets/HuggingFaceFW/fineweb"

def list_slice(dump: str, count: int) -> list[str]:
    """Return the first `count` parquet paths in `dump`, name-sorted (deterministic)."""
    hf = HfFileSystem()
    all_files = hf.glob(f"{HF_REPO}/data/{dump}/*.parquet")
    if not all_files:
        sys.exit(f"No parquet files found for dump {dump!r}")
    chosen = sorted(all_files)[:count]
    print(f"Selected {len(chosen)} / {len(all_files)} files from {dump}")
    return chosen

def stream_one(hf_path: str, bucket: str, dump: str, region: str) -> str:
    """Stream one file hf:// -> s3:// without touching local disk."""
    filename = hf_path.split("/")[-1]
    s3_key = f"fineweb/{dump}/{filename}"
    src = f"hf://{hf_path}"
    dst = f"s3://{bucket}/{s3_key}"

    # region in client_kwargs is REQUIRED (see below). Without it s3fs writes to
    # the wrong regional endpoint and PutObject fails: "No AWSAccessKey was presented".
    with fsspec.open(src, "rb") as fin, \
            fsspec.open(dst, "wb", client_kwargs={"region_name": region}) as fout:
        shutil.copyfileobj(fin, fout, length=32 * 1024 * 1024)  # 32MB chunks
    print(f"  uploaded {s3_key}")
    return s3_key

def main() -> None:
    p = argparse.ArgumentParser()
    p.add_argument("--dump", required=True)
    p.add_argument("--count", type=int, default=50)
    p.add_argument("--bucket", required=True)
    p.add_argument("--workers", type=int, default=8)
    p.add_argument("--region", default=os.environ.get("AWS_DEFAULT_REGION", "ap-northeast-1"))
    args = p.parse_args()

    files = list_slice(args.dump, args.count)

    # network I/O -> threads (not processes) overlap the waiting
    with concurrent.futures.ThreadPoolExecutor(max_workers=args.workers) as pool:
        futures = [pool.submit(stream_one, f, args.bucket, args.dump, args.region) for f in files]
        for fut in concurrent.futures.as_completed(futures):
            fut.result()

    print(f"Done: {len(files)} files -> s3://{args.bucket}/fineweb/{args.dump}/")

if __name__ == "__main__":
    main()

# build_manifest.py
import argparse
import json
import os

import boto3
import pyarrow.parquet as pq
import fsspec

def build(bucket, dump, revision, pulled_at, region):
    s3 = boto3.client("s3", region_name=region)
    paginator = s3.get_paginator("list_objects_v2")
    entries = []

    for page in paginator.paginate(Bucket=bucket, Prefix=f"fineweb/{dump}/"):
        for obj in page.get("Contents", []):
            key = obj["Key"]
            if not key.endswith(".parquet"):
                continue

            # read only the parquet footer for the row count (not the whole file)
            with fsspec.open(f"s3://{bucket}/{key}", "rb",
                             client_kwargs={"region_name": region}) as f:
                rows = pq.read_metadata(f).num_rows

            entries.append({
                "path": key,
                "size_bytes": obj["Size"],
                "etag": obj["ETag"].strip('"'),  # multipart ETags carry a "-N" suffix
                "rows": rows,
                "hf_revision": revision,
                "pulled_at": pulled_at,
            })
            print(f"  manifested {key} ({rows} rows)")

    entries.sort(key=lambda e: e["path"])
    return entries

def main():
    p = argparse.ArgumentParser()
    p.add_argument("--dump", required=True)
    p.add_argument("--bucket", required=True)
    p.add_argument("--revision", default="main")
    p.add_argument("--pulled-at", default="unknown")
    p.add_argument("--region", default=os.environ.get("AWS_DEFAULT_REGION", "ap-northeast-1"))
    args = p.parse_args()

    entries = build(args.bucket, args.dump, args.revision, args.pulled_at, args.region)
    if not entries:
        raise SystemExit(f"No parquet objects under fineweb/{args.dump}/")

    body = "\n".join(json.dumps(e) for e in entries) + "\n"
    manifest_key = f"manifests/{args.dump}-{len(entries)}files-v1.jsonl"

    boto3.client("s3", region_name=args.region).put_object(
        Bucket=args.bucket, Key=manifest_key,
        Body=body.encode("utf-8"), ContentType="application/x-ndjson",
    )
    print(f"Wrote manifest: s3://{args.bucket}/{manifest_key} ({len(entries)} entries)")

if __name__ == "__main__":
    main()

{"path": "fineweb/CC-MAIN-2024-10/000_00000.parquet", "size_bytes": 2147483648, "etag": "a4f3...-21", "rows": 973991, "hf_revision": "main", "pulled_at": "2026-07-25T09:12:33Z"}

# fineweb.yaml
resources:
  cloud: aws
  region: ap-northeast-1
  instance_type: c7g.2xlarge   # 16GB RAM (see the memory-shortage section below)
  use_spot: false              # on-demand (see the spot section below)
  disk_size: 30                # boot disk only; nothing lands on local disk

envs:
  HF_TOKEN: null                     # supplied via --env at launch
  S3_BUCKET: null                    # supplied via --env at launch
  CC_DUMP: CC-MAIN-2024-10
  FILE_COUNT: 50                     # ~2.0 GB/file measured, so 50 x 2.0 = ~100 GB
  AWS_DEFAULT_REGION: ap-northeast-1 # must match the bucket region (see below)

workdir: .

setup: |
  pip install 'huggingface_hub[hf_xet]>=1.5' 's3fs' 'pyarrow' 'boto3'

run: |
  set -e
  python stream_fineweb_slice.py --dump "$CC_DUMP" --count "$FILE_COUNT" --bucket "$S3_BUCKET" --workers 4
  python build_manifest.py --dump "$CC_DUMP" --bucket "$S3_BUCKET" \
      --revision main --pulled-at "$(date -u +%Y-%m-%dT%H:%M:%SZ)"

# .sky.yaml (placed in test/; auto-loaded when you run sky from that dir)
aws:
  remote_identity: SERVICE_ACCOUNT

cd test
uv run sky launch -c fw-pull -i 5 --down fineweb.yaml \
    --env HF_TOKEN=none \
    --env S3_BUCKET=$(terraform -chdir=infra output -raw bucket_name) \
    --env FILE_COUNT=50

# 1. Is the credential chain (IMDS/role) working at all?
python -c "import boto3; print(boto3.client('sts').get_caller_identity())"
# 2. Is an empty AWS_* env var breaking the chain?
env | grep -i aws
# 3. s3fs READ auth test
python -c "import s3fs; print(s3fs.S3FileSystem().ls(''))"
# 4. test WRITE separately (read working != write working)

  uploaded fineweb/CC-MAIN-2024-10/000_00049.parquet
Done: 50 files -> s3://<bucket>/fineweb/CC-MAIN-2024-10/
  manifested fineweb/CC-MAIN-2024-10/000_00049.parquet (957982 rows)
Wrote manifest: s3://<bucket>/manifests/CC-MAIN-2024-10-50files-v1.jsonl (50 entries)

head = s3.head_object(Bucket=bucket, Key=entry["path"])
assert head["ETag"].strip('"') == entry["etag"]
assert head["ContentLength"] == entry["size_bytes"]