diff --git a/pyproject.toml b/pyproject.toml index f3398a0322..c765fa7703 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -137,6 +137,12 @@ image_cpu = [ "torchvision" ] +image_lance = [ + "nemo_curator[image_cpu]", + "nemo_curator[lance]", + "s3fs>=2024.12.0", +] + # NVIDIA DALI (simplified; update the package to match your CUDA version if needed) image_cuda12 = [ "nemo_curator[image_cpu]", diff --git a/tests/stages/image/test_lance_writer_tutorial.py b/tests/stages/image/test_lance_writer_tutorial.py new file mode 100644 index 0000000000..4810fb5543 --- /dev/null +++ b/tests/stages/image/test_lance_writer_tutorial.py @@ -0,0 +1,129 @@ +# Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import io +import json +import tarfile +from pathlib import Path + +import pyarrow as pa +import pyarrow.parquet as pq +from PIL import Image + +from tutorials.image.lance_writer.manifest import PhysicalTar, build_manifest, load_manifest +from tutorials.image.lance_writer.pipeline import build_pipeline +from tutorials.image.lance_writer.stages import ( + IMAGE_SCHEMA, + candidate_is_better, + iter_tar_candidates, +) + + +def _inventory(path: Path, *, etag: str = "etag-a") -> None: + rows = [ + { + "source_shard": shard, + "tar_id": tar_id, + "attempt": attempt, + "tar_uri": f"s3://source/shard_{shard:05d}/attempt_{attempt}/{tar_id}.tar", + "tar_size": size, + "tar_etag": etag if tar_id == "00000" else f"etag-{tar_id}-{attempt}", + "last_modified": "2026-01-01T00:00:00+00:00", + } + for shard, tar_id, attempt, size in [ + (1, "00000", 1, 700), + (1, "00000", 2, 700), + (1, "00001", 1, 700), + (1, "00002", 1, 700), + (2, "00000", 1, 400), + ] + ] + pq.write_table(pa.Table.from_pylist(rows), path) + + +def test_manifest_is_stable_and_packs_never_cross_source_shards(tmp_path: Path) -> None: + inventory_path = tmp_path / "inventory.parquet" + manifest_dir = tmp_path / "manifest" + _inventory(inventory_path) + + first = build_manifest(str(inventory_path), str(manifest_dir), target_pack_bytes=1_000) + second = build_manifest(str(inventory_path), str(manifest_dir), target_pack_bytes=1_000) + metadata, packs = load_manifest(str(manifest_dir)) + + assert first == second == metadata + assert metadata["physical_tar_count"] == 5 + assert metadata["fpp_count"] == 4 + assert all({fpp.source_shard for fpp in pack.fpps} == {pack.source_shard} for pack in packs) + retried = next(fpp for pack in packs for fpp in pack.fpps if fpp.fpp_id == "shard_00001/00000") + assert [item.attempt for item in retried.attempts] == [1, 2] + + +def test_winner_policy_prefers_resolution_then_newest_attempt() -> None: + base = {"width": 100, "height": 100, "source_attempt": 1, "source_id": "b"} + assert candidate_is_better({**base, "width": 101}, base) + assert candidate_is_better({**base, "source_attempt": 2}, base) + assert candidate_is_better({**base, "source_id": "a"}, base) + assert not candidate_is_better({**base, "source_attempt": 0}, base) + + +def test_materializer_and_writer_are_fusible_one_cpu_stages() -> None: + pipeline = build_pipeline( + manifest_dir="/manifest", + dataset_uri="s3://output/images", + lance_commit_path="/checkpoints/lance", + source_storage_options={}, + lance_storage_options={}, + ) + + materializer, writer = pipeline.stages[1:] + assert materializer.resources == writer.resources + assert materializer.resources.cpus == 1 + assert pa.types.is_large_binary(IMAGE_SCHEMA.field("image").type) + + +def test_tar_reader_pairs_json_with_encoded_image_and_preserves_format() -> None: + image_buffer = io.BytesIO() + Image.new("RGB", (12, 8), color="red").save(image_buffer, format="JPEG") + metadata = json.dumps( + { + "status": "success", + "url": "https://example.test/image", + "width": 12, + "height": 8, + } + ).encode() + tar_buffer = io.BytesIO() + with tarfile.open(fileobj=tar_buffer, mode="w") as archive: + for name, payload in (("sample.json", metadata), ("sample.jpg", image_buffer.getvalue())): + info = tarfile.TarInfo(name) + info.size = len(payload) + archive.addfile(info, io.BytesIO(payload)) + tar_buffer.seek(0) + physical_tar = PhysicalTar( + source_shard=1, + tar_id="00000", + attempt=1, + tar_uri="s3://source/shard_00001/attempt_1/00000.tar", + tar_size=len(tar_buffer.getvalue()), + tar_etag="etag", + last_modified="2026-01-01T00:00:00+00:00", + ) + + rows = list(iter_tar_candidates(tar_buffer, physical_tar)) + + assert len(rows) == 1 + assert rows[0]["url"] == "https://example.test/image" + assert rows[0]["image_format"] == "JPEG" + assert rows[0]["mime_type"] == "image/jpeg" + assert (rows[0]["width"], rows[0]["height"]) == (12, 8) diff --git a/tutorials/image/README.md b/tutorials/image/README.md index 7f9462a85c..dfb952409c 100644 --- a/tutorials/image/README.md +++ b/tutorials/image/README.md @@ -11,6 +11,7 @@ Hands-on tutorials for curating image data with NeMo Curator. Complete working e | Tutorial | Description | Files | |----------|-------------|-------| | **[Getting Started](getting-started/)** | Image curation fundamentals | `image_curation_example.py`, `image_dedup_example.py`, `helper.py` | +| **[Lance Writer](lance_writer/)** | Retry-aware tar ingestion into URL-addressable Lance tables | `pipeline.py`, `manifest.py`, `stages.py`, `submit_array.sh` | ## Documentation Links @@ -23,4 +24,4 @@ Hands-on tutorials for curating image data with NeMo Curator. Complete working e ## Support -**Documentation**: [Main Docs](https://docs.nvidia.com/nemo/curator/latest/) • [API Reference](https://docs.nvidia.com/nemo/curator/latest/apidocs/index.html) • [GitHub Discussions](https://github.com/NVIDIA-NeMo/Curator/discussions) \ No newline at end of file +**Documentation**: [Main Docs](https://docs.nvidia.com/nemo/curator/latest/) • [API Reference](https://docs.nvidia.com/nemo/curator/latest/apidocs/index.html) • [GitHub Discussions](https://github.com/NVIDIA-NeMo/Curator/discussions) diff --git a/tutorials/image/lance_writer/README.md b/tutorials/image/lance_writer/README.md new file mode 100644 index 0000000000..c1feb9a2ad --- /dev/null +++ b/tutorials/image/lance_writer/README.md @@ -0,0 +1,159 @@ +# Retry-attempt image tars to Lance + +This workflow builds a URL-addressable Lance table from image/JSON pairs stored +in tar files. It is designed for sources that retry the same logical tar: every +physical attempt is processed together, and exactly one winner is selected per +exact URL before the image is written. + +The source contract must keep every retry copy of a URL in the same stable +`(source_shard, tar_id)` FPP. Cross-tar or cross-shard URL deduplication is a +separate dataset-wide operation and is not performed by this ingestion recipe. + +The example focuses on ingestion. Dataset-wide validation, compaction, +secondary-index construction, and benchmarking are intentionally outside its +scope. + +## Pipeline + +```text +immutable tar inventory + │ + ▼ +stable (source_shard, tar_id) FPPs + │ group all attempts; pack within one source shard toward 1 GiB + ▼ +FppPackPartitioningStage + │ deterministic task IDs + Slurm-array filtering + ▼ +FppPackMaterializationStage ──fused──> LanceWriter + │ │ + │ Arrow large_binary images └─ uncommitted fragments + checkpoints + ▼ +commit_lance_checkpoint (once, after every logical shard is complete) +``` + +The materializer and writer both request one CPU, allowing Ray Data to fuse +them. This avoids retaining an additional image-heavy intermediate block. + +## Input contracts + +The inventory is a Parquet table with one row per physical tar: + +| Column | Type | Meaning | +|---|---|---| +| `source_shard` | integer | Source shard containing the logical tar | +| `tar_id` | string | Stable logical tar ID | +| `attempt` | integer | Monotonically increasing retry attempt | +| `tar_uri` | string | Full `s3://` URI | +| `tar_size` | integer | Object size in bytes | +| `tar_etag` | string | Object ETag | +| `last_modified` | string | Stable timestamp from the inventory pass | + +Each tar contains image and JSON members with the same member stem. JSON rows +must contain `url`; `width`, `height`, `sha256`, and `status` are consumed when +present. Pillow determines the actual encoded format, so JPEG, MPO, PNG, and +WebP payloads remain distinguishable even when the member suffix is `.jpg`. + +For equal URLs, the winner policy is: + +1. largest pixel area; +2. newest attempt; +3. lexicographically smallest source ID as a deterministic tie-breaker. + +The output includes the encoded image as ordinary Arrow `large_binary`, MD5 +and SHA-256 hashes, image format/MIME/dimensions, the original JSON, and only +the winning tar/member provenance. + +## Environment + +Install the image and Lance dependencies directly: + +```bash +pip install -e '.[image_lance]' +``` + +Use the standard AWS environment variables for credentials. S3-compatible +endpoint settings are explicit JSON objects because fsspec and Lance use +different option shapes: + +```bash +export SOURCE_STORAGE_OPTIONS='{"client_kwargs":{"endpoint_url":"https://s3.example"}}' +export LANCE_STORAGE_OPTIONS='{"endpoint":"https://s3.example","aws_region":"us-east-1","virtual_hosted_style_request":"false","client_max_retries":"20"}' +``` + +## Build the manifest once + +Run this on a CPU node after the source inventory is frozen: + +```bash +python -m tutorials.image.lance_writer.pipeline build-manifest \ + --inventory /shared/inventory/physical_tars.parquet \ + --manifest-dir /shared/manifests/images-v1 \ + --target-pack-bytes 1073741824 +``` + +The snapshot ID hashes the inventory descriptors and pack target. The command +reuses an identical manifest and refuses to overwrite a different one. + +## Run with Slurm arrays and resumability + +Choose enough logical shards to give each array element many packs. Limit +concurrent nodes from measured object-store throughput rather than CPU count. + +```bash +export MANIFEST_DIR=/shared/manifests/images-v1 +export DATASET_URI=s3://output-bucket/lance/images-v1 +export LANCE_COMMIT_PATH=/shared/checkpoints/images-v1/lance +export CHECKPOINT_PATH=/shared/checkpoints/images-v1/curator + +sbatch --account= --partition= \ + --array=0-499%60 --export=ALL \ + tutorials/image/lance_writer/submit_array.sh +``` + +Both checkpoint paths should be on shared storage. Dataset payload is written +only to Lance; source tars are read through fsspec and are never copied to an +intermediate object-store prefix. + +After the array finishes, use Curator's Slurm retry helper with the same +`CHECKPOINT_PATH`. The `fields` format preserves the original logical shard +count when the physical retry array is sparse: + +```bash +python tutorials/slurm/retry_array.py \ + --checkpoint-path "${CHECKPOINT_PATH}" \ + --format fields \ + --max-array-size 1001 +``` + +Each output line contains `array_expression shard_index_offset +minimum_shard_index original_total_shards`. Resubmit only those indices while +preserving the logical values: + +```bash +while read -r array offset minimum total; do + sbatch --account= --partition= \ + --array="${array}%60" \ + --export="ALL,SHARD_INDEX_OFFSET=${offset},MINIMUM_SHARD_INDEX=${minimum},TOTAL_SHARDS=${total}" \ + tutorials/image/lance_writer/submit_array.sh +done < <( + python tutorials/slurm/retry_array.py \ + --checkpoint-path "${CHECKPOINT_PATH}" \ + --format fields \ + --max-array-size 1001 +) +``` + +Repeat until the helper prints no lines, then atomically commit all +checkpointed fragments: + +```bash +python -m tutorials.image.lance_writer.pipeline commit \ + --dataset-uri "${DATASET_URI}" \ + --lance-commit-path "${LANCE_COMMIT_PATH}" \ + --lance-storage-options "${LANCE_STORAGE_OPTIONS}" +``` + +Do not run the commit while array shards are still writing. `LanceWriter` +checkpoint records make individual packs idempotent; Curator's completion +manifests determine which logical array shards need another attempt. diff --git a/tutorials/image/lance_writer/__init__.py b/tutorials/image/lance_writer/__init__.py new file mode 100644 index 0000000000..bb4a5cf320 --- /dev/null +++ b/tutorials/image/lance_writer/__init__.py @@ -0,0 +1 @@ +"""Build a URL-addressable Lance image table from retry-attempt tar shards.""" diff --git a/tutorials/image/lance_writer/manifest.py b/tutorials/image/lance_writer/manifest.py new file mode 100644 index 0000000000..28195b2dae --- /dev/null +++ b/tutorials/image/lance_writer/manifest.py @@ -0,0 +1,318 @@ +# Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Create and read the immutable input manifest used by the Lance tutorial.""" + +from __future__ import annotations + +import hashlib +import json +from collections import defaultdict +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any + +import pyarrow as pa +import pyarrow.parquet as pq + +MANIFEST_SCHEMA_VERSION = 1 +PACKS_FILENAME = "fpp_packs.parquet" +METADATA_FILENAME = "manifest.json" + + +@dataclass(frozen=True) +class PhysicalTar: + """One physical attempt for a stable ``(source_shard, tar_id)`` unit.""" + + source_shard: int + tar_id: str + attempt: int + tar_uri: str + tar_size: int + tar_etag: str + last_modified: str + + @property + def descriptor(self) -> str: + return "\0".join((self.tar_uri, str(self.tar_size), self.tar_etag, self.last_modified)) + + +@dataclass(frozen=True) +class FppPartition: + """All attempts for one stable ``(source_shard, tar_id)`` partition.""" + + snapshot_id: str + fpp_id: str + source_shard: int + tar_id: str + attempts: tuple[PhysicalTar, ...] + total_tar_bytes: int + max_tar_bytes: int + + @property + def deterministic_id(self) -> str: + digest = hashlib.sha256(f"{self.snapshot_id}\0{self.fpp_id}\n".encode()) + for item in self.attempts: + digest.update(f"{item.attempt}\0{item.descriptor}\n".encode()) + return digest.hexdigest()[:32] + + +@dataclass(frozen=True) +class FppPack: + """A deterministic within-shard group targeting one Lance fragment.""" + + snapshot_id: str + pack_id: str + source_shard: int + fpps: tuple[FppPartition, ...] + estimated_output_bytes: int + total_tar_bytes: int + + @property + def deterministic_id(self) -> str: + digest = hashlib.sha256(f"{self.snapshot_id}\0{self.pack_id}\n".encode()) + for fpp in self.fpps: + digest.update(f"{fpp.deterministic_id}\n".encode()) + return digest.hexdigest()[:32] + + +PHYSICAL_TAR_SCHEMA = pa.struct( + [ + pa.field("source_shard", pa.int32(), nullable=False), + pa.field("tar_id", pa.string(), nullable=False), + pa.field("attempt", pa.int16(), nullable=False), + pa.field("tar_uri", pa.string(), nullable=False), + pa.field("tar_size", pa.int64(), nullable=False), + pa.field("tar_etag", pa.string(), nullable=False), + pa.field("last_modified", pa.string(), nullable=False), + ] +) +FPP_SCHEMA = pa.struct( + [ + pa.field("snapshot_id", pa.string(), nullable=False), + pa.field("fpp_id", pa.string(), nullable=False), + pa.field("source_shard", pa.int32(), nullable=False), + pa.field("tar_id", pa.string(), nullable=False), + pa.field("attempts", pa.list_(PHYSICAL_TAR_SCHEMA), nullable=False), + pa.field("total_tar_bytes", pa.int64(), nullable=False), + pa.field("max_tar_bytes", pa.int64(), nullable=False), + ] +) +PACK_SCHEMA = pa.schema( + [ + pa.field("snapshot_id", pa.string(), nullable=False), + pa.field("pack_id", pa.string(), nullable=False), + pa.field("source_shard", pa.int32(), nullable=False), + pa.field("fpps", pa.list_(FPP_SCHEMA), nullable=False), + pa.field("estimated_output_bytes", pa.int64(), nullable=False), + pa.field("total_tar_bytes", pa.int64(), nullable=False), + ] +) + +INVENTORY_COLUMNS = { + "source_shard", + "tar_id", + "attempt", + "tar_uri", + "tar_size", + "tar_etag", + "last_modified", +} + + +def _snapshot_id(rows: list[PhysicalTar], target_pack_bytes: int) -> str: + digest = hashlib.sha256(f"schema={MANIFEST_SCHEMA_VERSION}\0target_pack_bytes={target_pack_bytes}\n".encode()) + for row in sorted(rows, key=lambda item: item.tar_uri): + digest.update(row.descriptor.encode()) + digest.update(b"\n") + return digest.hexdigest()[:24] + + +def _validated_inventory(inventory_path: str) -> list[PhysicalTar]: + table = pq.read_table(inventory_path) + missing = INVENTORY_COLUMNS - set(table.column_names) + if missing: + message = f"Inventory is missing required columns: {sorted(missing)}" + raise ValueError(message) + rows = [PhysicalTar(**row) for row in table.select(sorted(INVENTORY_COLUMNS)).to_pylist()] + if not rows: + message = "Inventory contains no tar objects" + raise ValueError(message) + + identities: set[tuple[int, str, int]] = set() + uris: set[str] = set() + for row in rows: + identity = (row.source_shard, row.tar_id, row.attempt) + if identity in identities: + message = f"Duplicate physical tar identity: {identity}" + raise ValueError(message) + if row.tar_uri in uris: + message = f"Duplicate physical tar URI: {row.tar_uri}" + raise ValueError(message) + if row.tar_size <= 0: + message = f"Tar has non-positive size: {row.tar_uri}" + raise ValueError(message) + identities.add(identity) + uris.add(row.tar_uri) + return sorted(rows, key=lambda item: (item.source_shard, item.tar_id, item.attempt)) + + +def build_fpp_partitions(physical_tars: list[PhysicalTar], snapshot_id: str) -> list[FppPartition]: + grouped: dict[tuple[int, str], list[PhysicalTar]] = defaultdict(list) + for physical_tar in physical_tars: + grouped[(physical_tar.source_shard, physical_tar.tar_id)].append(physical_tar) + + partitions = [] + for (source_shard, tar_id), attempts in sorted(grouped.items()): + ordered = tuple(sorted(attempts, key=lambda item: item.attempt)) + partitions.append( + FppPartition( + snapshot_id=snapshot_id, + fpp_id=f"shard_{source_shard:05d}/{tar_id}", + source_shard=source_shard, + tar_id=tar_id, + attempts=ordered, + total_tar_bytes=sum(item.tar_size for item in ordered), + max_tar_bytes=max(item.tar_size for item in ordered), + ) + ) + return partitions + + +def build_fpp_packs(partitions: list[FppPartition], target_pack_bytes: int) -> list[FppPack]: + """Balance indivisible FPPs within each source shard toward a byte target.""" + if target_pack_bytes <= 0: + message = "target_pack_bytes must be positive" + raise ValueError(message) + by_shard: dict[int, list[FppPartition]] = defaultdict(list) + for partition in partitions: + by_shard[partition.source_shard].append(partition) + + packs = [] + for source_shard, shard_fpps in sorted(by_shard.items()): + estimated_total = sum(fpp.max_tar_bytes for fpp in shard_fpps) + pack_count = max(1, int(estimated_total / target_pack_bytes + 0.5)) + pack_count = min(pack_count, len(shard_fpps)) + bins: list[list[FppPartition]] = [[] for _ in range(pack_count)] + bin_bytes = [0] * pack_count + for fpp in sorted(shard_fpps, key=lambda item: (-item.max_tar_bytes, item.tar_id)): + bin_index = min(range(pack_count), key=lambda index: (bin_bytes[index], index)) + bins[bin_index].append(fpp) + bin_bytes[bin_index] += fpp.max_tar_bytes + + ordered_bins = sorted( + (tuple(sorted(items, key=lambda item: item.tar_id)) for items in bins if items), + key=lambda items: items[0].tar_id, + ) + for index, items in enumerate(ordered_bins): + packs.append( + FppPack( + snapshot_id=items[0].snapshot_id, + pack_id=f"shard_{source_shard:05d}/pack_{index:03d}", + source_shard=source_shard, + fpps=items, + estimated_output_bytes=sum(item.max_tar_bytes for item in items), + total_tar_bytes=sum(item.total_tar_bytes for item in items), + ) + ) + return packs + + +def _fpp_dict(fpp: FppPartition) -> dict[str, Any]: + return { + **asdict(fpp), + "attempts": [asdict(attempt) for attempt in fpp.attempts], + } + + +def _pack_dict(pack: FppPack) -> dict[str, Any]: + return { + **asdict(pack), + "fpps": [_fpp_dict(fpp) for fpp in pack.fpps], + } + + +def build_manifest(inventory_path: str, manifest_dir: str, target_pack_bytes: int) -> dict[str, Any]: + """Build the manifest once; refuse to overwrite a different snapshot.""" + physical_tars = _validated_inventory(inventory_path) + snapshot_id = _snapshot_id(physical_tars, target_pack_bytes) + partitions = build_fpp_partitions(physical_tars, snapshot_id) + packs = build_fpp_packs(partitions, target_pack_bytes) + metadata = { + "manifest_schema_version": MANIFEST_SCHEMA_VERSION, + "snapshot_id": snapshot_id, + "target_pack_bytes": target_pack_bytes, + "physical_tar_count": len(physical_tars), + "fpp_count": len(partitions), + "fpp_pack_count": len(packs), + "source_shard_count": len({item.source_shard for item in physical_tars}), + "physical_tar_bytes": sum(item.tar_size for item in physical_tars), + } + + root = Path(manifest_dir) + metadata_path = root / METADATA_FILENAME + packs_path = root / PACKS_FILENAME + if metadata_path.exists() or packs_path.exists(): + if metadata_path.exists() and packs_path.exists(): + existing = json.loads(metadata_path.read_text()) + if existing == metadata: + return existing + message = f"Refusing to overwrite a different manifest in {root}" + raise ValueError(message) + + root.mkdir(parents=True, exist_ok=True) + temporary_packs = packs_path.with_suffix(".parquet.tmp") + pq.write_table(pa.Table.from_pylist([_pack_dict(pack) for pack in packs], schema=PACK_SCHEMA), temporary_packs) + temporary_packs.replace(packs_path) + temporary_metadata = metadata_path.with_suffix(".json.tmp") + temporary_metadata.write_text(json.dumps(metadata, indent=2, sort_keys=True) + "\n") + temporary_metadata.replace(metadata_path) + return metadata + + +def _physical_tar(row: dict[str, Any]) -> PhysicalTar: + return PhysicalTar(**row) + + +def _fpp(row: dict[str, Any]) -> FppPartition: + return FppPartition( + snapshot_id=str(row["snapshot_id"]), + fpp_id=str(row["fpp_id"]), + source_shard=int(row["source_shard"]), + tar_id=str(row["tar_id"]), + attempts=tuple(_physical_tar(item) for item in row["attempts"]), + total_tar_bytes=int(row["total_tar_bytes"]), + max_tar_bytes=int(row["max_tar_bytes"]), + ) + + +def load_manifest(manifest_dir: str) -> tuple[dict[str, Any], list[FppPack]]: + root = Path(manifest_dir) + metadata = json.loads((root / METADATA_FILENAME).read_text()) + rows = pq.read_table(root / PACKS_FILENAME).to_pylist() + packs = [ + FppPack( + snapshot_id=str(row["snapshot_id"]), + pack_id=str(row["pack_id"]), + source_shard=int(row["source_shard"]), + fpps=tuple(_fpp(item) for item in row["fpps"]), + estimated_output_bytes=int(row["estimated_output_bytes"]), + total_tar_bytes=int(row["total_tar_bytes"]), + ) + for row in rows + ] + if {pack.snapshot_id for pack in packs} != {metadata["snapshot_id"]}: + message = "Manifest metadata and FPP packs have different snapshot IDs" + raise ValueError(message) + return metadata, packs diff --git a/tutorials/image/lance_writer/pipeline.py b/tutorials/image/lance_writer/pipeline.py new file mode 100644 index 0000000000..d7e3bce21c --- /dev/null +++ b/tutorials/image/lance_writer/pipeline.py @@ -0,0 +1,169 @@ +# Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Build, ingest, and commit the tar-attempt-to-Lance image workflow.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path +from typing import Any + +from loguru import logger + +from nemo_curator.backends.ray_data import RayDataExecutor +from nemo_curator.core.client import RayClient +from nemo_curator.pipeline import Pipeline +from nemo_curator.stages.resources import Resources +from nemo_curator.stages.text.io.writer import LanceWriter, commit_lance_checkpoint + +from .manifest import METADATA_FILENAME, build_manifest +from .stages import ( + IMAGE_SCHEMA, + FppPackMaterializationStage, + FppPackPartitioningStage, +) + + +def _json_object(value: str) -> dict[str, Any]: + parsed = json.loads(value) + if not isinstance(parsed, dict): + message = "storage options must be a JSON object" + raise argparse.ArgumentTypeError(message) + return parsed + + +def _canonical_json(value: dict[str, Any]) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":")) + + +def build_pipeline( + *, + manifest_dir: str, + dataset_uri: str, + lance_commit_path: str, + source_storage_options: dict[str, Any], + lance_storage_options: dict[str, Any], +) -> Pipeline: + """Compose a source, fused materializer, and checkpointed Lance sink.""" + writer = LanceWriter( + path=dataset_uri, + commit_path=lance_commit_path, + schema=IMAGE_SCHEMA, + mode="create", + write_kwargs={ + "storage_options": lance_storage_options, + "data_storage_version": "2.2", + "max_rows_per_file": 500_000, + }, + ).with_(resources=Resources(cpus=1), batch_size=1) + writer.is_sink_stage = True + return Pipeline( + name="tar_attempts_to_lance_images", + description="Stable FPP retry-attempt deduplication and large-binary Lance writing", + stages=[ + FppPackPartitioningStage( + manifest_dir=manifest_dir, + dataset_name=dataset_uri, + ), + FppPackMaterializationStage(source_storage_options_json=_canonical_json(source_storage_options)), + writer, + ], + ) + + +def _ingest(args: argparse.Namespace) -> None: + metadata = json.loads((Path(args.manifest_dir) / METADATA_FILENAME).read_text()) + pipeline = build_pipeline( + manifest_dir=args.manifest_dir, + dataset_uri=args.dataset_uri, + lance_commit_path=args.lance_commit_path, + source_storage_options=args.source_storage_options, + lance_storage_options=args.lance_storage_options, + ) + logger.info("Snapshot: {}\n{}", metadata["snapshot_id"], pipeline.describe()) + ray_kwargs = {"ray_temp_dir": args.ray_temp_dir} if args.ray_temp_dir else {} + ray_client = RayClient( + num_cpus=args.cpus, + object_store_memory=args.object_store_memory, + include_dashboard=False, + **ray_kwargs, + ) + try: + ray_client.start() + pipeline.run( + executor=RayDataExecutor(), + checkpoint_path=args.checkpoint_path, + ) + finally: + ray_client.stop() + + +def _commit(args: argparse.Namespace) -> None: + version = commit_lance_checkpoint( + args.dataset_uri, + args.lance_commit_path, + storage_options=args.lance_storage_options, + ) + print(json.dumps({"dataset_uri": args.dataset_uri, "version": version}, indent=2)) + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Build a URL-addressable Lance image table from tar attempts", + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + ) + subparsers = parser.add_subparsers(dest="command", required=True) + + manifest_parser = subparsers.add_parser("build-manifest") + manifest_parser.add_argument("--inventory", required=True) + manifest_parser.add_argument("--manifest-dir", required=True) + manifest_parser.add_argument("--target-pack-bytes", type=int, default=1024**3) + + ingest_parser = subparsers.add_parser("ingest") + ingest_parser.add_argument("--manifest-dir", required=True) + ingest_parser.add_argument("--dataset-uri", required=True) + ingest_parser.add_argument("--lance-commit-path", required=True) + ingest_parser.add_argument("--checkpoint-path", required=True) + ingest_parser.add_argument("--source-storage-options", type=_json_object, default={}) + ingest_parser.add_argument("--lance-storage-options", type=_json_object, default={}) + ingest_parser.add_argument("--cpus", type=int, default=8) + ingest_parser.add_argument("--object-store-memory", type=int, default=None) + ingest_parser.add_argument("--ray-temp-dir", default=None) + + commit_parser = subparsers.add_parser("commit") + commit_parser.add_argument("--dataset-uri", required=True) + commit_parser.add_argument("--lance-commit-path", required=True) + commit_parser.add_argument("--lance-storage-options", type=_json_object, default={}) + return parser + + +def main() -> None: + args = _parser().parse_args() + if args.command == "build-manifest": + result = build_manifest( + args.inventory, + args.manifest_dir, + args.target_pack_bytes, + ) + print(json.dumps(result, indent=2, sort_keys=True)) + elif args.command == "ingest": + _ingest(args) + else: + _commit(args) + + +if __name__ == "__main__": + main() diff --git a/tutorials/image/lance_writer/stages.py b/tutorials/image/lance_writer/stages.py new file mode 100644 index 0000000000..b770a40d49 --- /dev/null +++ b/tutorials/image/lance_writer/stages.py @@ -0,0 +1,321 @@ +# Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Custom source and materialization stages for the Lance image workflow.""" + +from __future__ import annotations + +import hashlib +import io +import json +import tarfile +import time +from dataclasses import dataclass, field +from functools import lru_cache +from pathlib import PurePosixPath +from typing import TYPE_CHECKING, Any, BinaryIO + +import fsspec +import pyarrow as pa +from loguru import logger +from PIL import Image + +from nemo_curator.backends.utils import RayStageSpecKeys +from nemo_curator.stages.base import ProcessingStage +from nemo_curator.stages.resources import Resources +from nemo_curator.tasks import DocumentBatch, EmptyTask, Task + +from .manifest import FppPack, PhysicalTar, load_manifest + +if TYPE_CHECKING: + from collections.abc import Iterator + + from fsspec.spec import AbstractFileSystem + +IMAGE_SUFFIXES = frozenset({".jpg", ".jpeg", ".mpo", ".png", ".webp"}) + +# The stage verifies encoded images without decoding their raster into memory. +Image.MAX_IMAGE_PIXELS = None + +IMAGE_SCHEMA = pa.schema( + [ + pa.field("url", pa.string(), nullable=False), + pa.field("image", pa.large_binary(), nullable=False), + pa.field("image_format", pa.string(), nullable=False), + pa.field("mime_type", pa.string(), nullable=False), + pa.field("image_size_bytes", pa.int64(), nullable=False), + pa.field("md5", pa.string(), nullable=False), + pa.field("sha256", pa.string(), nullable=False), + pa.field("source_sha256", pa.string(), nullable=True), + pa.field("source_sha256_matches", pa.bool_(), nullable=True), + pa.field("width", pa.int32(), nullable=False), + pa.field("height", pa.int32(), nullable=False), + pa.field("source_shard", pa.int32(), nullable=False), + pa.field("source_attempt", pa.int16(), nullable=False), + pa.field("source_tar_id", pa.string(), nullable=False), + pa.field("source_tar_uri", pa.string(), nullable=False), + pa.field("source_image_member", pa.string(), nullable=False), + pa.field("source_json_member", pa.string(), nullable=False), + pa.field("metadata_json", pa.large_string(), nullable=False), + ] +) + + +@dataclass +class FppPackTask(Task[FppPack]): + """One deterministic, within-shard pack and its complete retry history.""" + + @property + def num_items(self) -> int: + return sum(len(fpp.attempts) for fpp in self.data.fpps) + + def validate(self) -> bool: + if not self.data.fpps: + message = f"FPP pack {self.data.pack_id} is empty" + raise ValueError(message) + for fpp in self.data.fpps: + if fpp.source_shard != self.data.source_shard: + message = f"FPP pack {self.data.pack_id} crosses source shards" + raise ValueError(message) + attempts = [item.attempt for item in fpp.attempts] + if attempts != sorted(attempts) or len(attempts) != len(set(attempts)): + message = f"FPP {fpp.fpp_id} attempts are not sorted and unique" + raise ValueError(message) + return True + + def get_deterministic_id(self) -> str: + return self.data.deterministic_id + + +@dataclass +class FppPackPartitioningStage(ProcessingStage[EmptyTask, FppPackTask]): + """Emit the frozen FPP packs; Curator applies logical Slurm-array sharding.""" + + manifest_dir: str + dataset_name: str + name: str = "fpp_pack_partitioning" + is_source_stage: bool = True + resources: Resources = field(default_factory=lambda: Resources(cpus=0.5)) + + def ray_stage_spec(self) -> dict[str, Any]: + return {RayStageSpecKeys.IS_FANOUT_STAGE: True} + + def num_workers(self) -> int: + return 1 + + def process(self, _: EmptyTask) -> list[FppPackTask]: + metadata, packs = load_manifest(self.manifest_dir) + logger.info( + "Loaded snapshot {} with {:,} FPP packs", + metadata["snapshot_id"], + len(packs), + ) + return [FppPackTask(dataset_name=self.dataset_name, data=pack) for pack in packs] + + +@lru_cache(maxsize=8) +def _s3_filesystem(storage_options_json: str) -> AbstractFileSystem: + return fsspec.filesystem("s3", **json.loads(storage_options_json)) + + +def _member_key(member_name: str) -> tuple[str, str] | None: + suffix = PurePosixPath(member_name).suffix.lower() + if suffix == ".json": + return member_name[: -len(suffix)], "json" + if suffix in IMAGE_SUFFIXES: + return member_name[: -len(suffix)], "image" + return None + + +def _image_properties(image_bytes: bytes, source_id: str) -> tuple[str, str, int, int]: + try: + with Image.open(io.BytesIO(image_bytes)) as image: + image_format = image.format + mime_type = image.get_format_mimetype() + width, height = image.size + if not image_format or not mime_type: + _raise_missing_image_type() + image.verify() + except Exception as error: + message = f"Invalid encoded image {source_id}: {error}" + raise ValueError(message) from error + return image_format, mime_type, int(width), int(height) + + +def _raise_missing_image_type() -> None: + message = "Pillow did not identify the format and MIME type" + raise ValueError(message) + + +def _candidate( + physical_tar: PhysicalTar, + image_member: str, + json_member: str, + image_bytes: bytes, + metadata: dict[str, Any], +) -> dict[str, Any]: + url = metadata.get("url") + if not isinstance(url, str) or not url: + message = f"Successful record has no URL: {physical_tar.tar_uri}#{json_member}" + raise ValueError(message) + source_id = f"{physical_tar.tar_uri}#{image_member}" + image_format, mime_type, width, height = _image_properties(image_bytes, source_id) + if metadata.get("width") is not None and metadata.get("height") is not None: + expected_size = int(metadata["width"]), int(metadata["height"]) + if (width, height) != expected_size: + message = f"Image/JSON resolution mismatch in {source_id}: {(width, height)} != {expected_size}" + raise ValueError(message) + return { + "url": url, + "image": image_bytes, + "image_format": image_format, + "mime_type": mime_type, + "width": width, + "height": height, + "source_sha256": metadata.get("sha256"), + "source_id": source_id, + "source_shard": physical_tar.source_shard, + "source_attempt": physical_tar.attempt, + "source_tar_id": physical_tar.tar_id, + "source_tar_uri": physical_tar.tar_uri, + "source_image_member": image_member, + "source_json_member": json_member, + "metadata_json": json.dumps(metadata, sort_keys=True, separators=(",", ":"), ensure_ascii=False), + } + + +def iter_tar_candidates(fileobj: BinaryIO, physical_tar: PhysicalTar) -> Iterator[dict[str, Any]]: + """Pair image/JSON members from one streaming tar without temporary files.""" + pending_images: dict[str, tuple[str, bytes]] = {} + pending_json: dict[str, tuple[str, dict[str, Any]]] = {} + seen_members: set[str] = set() + with tarfile.open(fileobj=fileobj, mode="r|*") as archive: + for member in archive: + member_type = _member_key(member.name) + if not member.isfile() or member_type is None: + continue + if member.name in seen_members: + message = f"Duplicate tar member {physical_tar.tar_uri}#{member.name}" + raise ValueError(message) + seen_members.add(member.name) + stream = archive.extractfile(member) + if stream is None: + message = f"Unable to extract {physical_tar.tar_uri}#{member.name}" + raise ValueError(message) + key, kind = member_type + if kind == "image": + pending_images[key] = member.name, stream.read() + else: + try: + metadata = json.loads(stream.read()) + except (UnicodeDecodeError, json.JSONDecodeError): + continue + if not isinstance(metadata, dict) or metadata.get("status") not in (None, "success"): + continue + pending_json[key] = member.name, metadata + + if key in pending_images and key in pending_json: + image_member, image_bytes = pending_images.pop(key) + json_member, metadata = pending_json.pop(key) + try: + yield _candidate( + physical_tar, + image_member, + json_member, + image_bytes, + metadata, + ) + except (KeyError, TypeError, ValueError): + continue + + +def candidate_is_better(candidate: dict[str, Any], incumbent: dict[str, Any]) -> bool: + """Prefer larger resolution, then newer attempt, then stable source ID.""" + candidate_area = int(candidate["width"]) * int(candidate["height"]) + incumbent_area = int(incumbent["width"]) * int(incumbent["height"]) + if candidate_area != incumbent_area: + return candidate_area > incumbent_area + if int(candidate["source_attempt"]) != int(incumbent["source_attempt"]): + return int(candidate["source_attempt"]) > int(incumbent["source_attempt"]) + return str(candidate["source_id"]) < str(incumbent["source_id"]) + + +def _finalize(candidate: dict[str, Any]) -> dict[str, Any]: + row = {key: value for key, value in candidate.items() if key != "source_id"} + image_bytes = row["image"] + sha256 = hashlib.sha256(image_bytes).hexdigest() + source_sha256 = row["source_sha256"] + row.update( + { + "image_size_bytes": len(image_bytes), + "md5": hashlib.md5(image_bytes, usedforsecurity=False).hexdigest(), + "sha256": sha256, + "source_sha256_matches": (None if source_sha256 is None else sha256 == source_sha256), + } + ) + return row + + +@dataclass +class FppPackMaterializationStage(ProcessingStage[FppPackTask, DocumentBatch]): + """Read all attempts, choose one exact-URL winner, and emit Arrow large binary.""" + + source_storage_options_json: str = "{}" + name: str = "fpp_pack_materialization" + resources: Resources = field(default_factory=lambda: Resources(cpus=1)) + + def process(self, task: FppPackTask) -> DocumentBatch: + started = time.perf_counter() + filesystem = _s3_filesystem(self.source_storage_options_json) + winners: dict[str, dict[str, Any]] = {} + candidate_count = 0 + for fpp in task.data.fpps: + for physical_tar in fpp.attempts: + tar_bytes = filesystem.cat_ranges( + [physical_tar.tar_uri], + [0], + [physical_tar.tar_size], + on_error="raise", + )[0] + if len(tar_bytes) != physical_tar.tar_size: + message = f"Short read for {physical_tar.tar_uri}: {len(tar_bytes)} != {physical_tar.tar_size}" + raise ValueError(message) + for candidate in iter_tar_candidates(io.BytesIO(tar_bytes), physical_tar): + candidate_count += 1 + url = str(candidate["url"]) + incumbent = winners.get(url) + if incumbent is None or candidate_is_better(candidate, incumbent): + winners[url] = candidate + del tar_bytes + + rows = [_finalize(winners[url]) for url in sorted(winners)] + table = pa.Table.from_pylist(rows, schema=IMAGE_SCHEMA) + metrics = { + "snapshot_id": task.data.snapshot_id, + "pack_id": task.data.pack_id, + "source_shard": task.data.source_shard, + "fpp_count": len(task.data.fpps), + "physical_attempts": sum(len(fpp.attempts) for fpp in task.data.fpps), + "candidate_rows": candidate_count, + "winner_rows": len(rows), + "image_payload_bytes": sum(row["image_size_bytes"] for row in rows), + "arrow_bytes": table.nbytes, + "elapsed_seconds": time.perf_counter() - started, + } + logger.info("Materialized {}: {}", task.data.pack_id, metrics) + return DocumentBatch( + dataset_name=task.dataset_name, + data=table, + _metadata={"fpp_pack": metrics}, + ) diff --git a/tutorials/image/lance_writer/submit_array.sh b/tutorials/image/lance_writer/submit_array.sh new file mode 100755 index 0000000000..35459dd60d --- /dev/null +++ b/tutorials/image/lance_writer/submit_array.sh @@ -0,0 +1,50 @@ +#!/bin/bash +# One-node Ray cluster per logical Slurm-array shard. Pass site-specific +# --account and --partition options to sbatch rather than hard-coding them here. + +#SBATCH --job-name=image-lance +#SBATCH --nodes=1 +#SBATCH --ntasks=1 +#SBATCH --cpus-per-task=32 +#SBATCH --mem=0 +#SBATCH --exclusive +#SBATCH --time=04:00:00 +#SBATCH --output=image_lance_%A_%a.log +#SBATCH --error=image_lance_%A_%a.log + +set -euo pipefail + +: "${MANIFEST_DIR:?Set MANIFEST_DIR to the frozen manifest directory}" +: "${DATASET_URI:?Set DATASET_URI to the output Lance URI}" +: "${LANCE_COMMIT_PATH:?Set LANCE_COMMIT_PATH to shared fragment checkpoint storage}" +: "${CHECKPOINT_PATH:?Set CHECKPOINT_PATH to shared Curator checkpoint storage}" +: "${SLURM_ARRAY_TASK_ID:?Submit this script with sbatch --array}" +: "${SLURM_ARRAY_TASK_COUNT:?Submit this script with sbatch --array}" + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +CURATOR_DIR="${CURATOR_DIR:-$(cd "${SCRIPT_DIR}/../../.." && pwd)}" +SOURCE_STORAGE_OPTIONS="${SOURCE_STORAGE_OPTIONS:-{}}" +LANCE_STORAGE_OPTIONS="${LANCE_STORAGE_OPTIONS:-{}}" +CPUS="${SLURM_CPUS_PER_TASK:-32}" +SHARD_INDEX_OFFSET="${SHARD_INDEX_OFFSET:-0}" +SHARD_INDEX="${SHARD_INDEX:-$((SLURM_ARRAY_TASK_ID + SHARD_INDEX_OFFSET))}" +TOTAL_SHARDS="${TOTAL_SHARDS:-${SLURM_ARRAY_TASK_COUNT}}" +MINIMUM_SHARD_INDEX="${MINIMUM_SHARD_INDEX:-0}" + +export NEMO_CURATOR_SLURM_ARRAY_ENABLED=1 +export NEMO_CURATOR_SLURM_ARRAY_SHARD_INDEX="${SHARD_INDEX}" +export NEMO_CURATOR_SLURM_ARRAY_TOTAL_SHARDS="${TOTAL_SHARDS}" +export NEMO_CURATOR_SLURM_ARRAY_MINIMUM_SHARD_INDEX="${MINIMUM_SHARD_INDEX}" +export RAY_TMPDIR="/tmp/ray_${SLURM_ARRAY_JOB_ID}_${SLURM_ARRAY_TASK_ID}" +export PYTHONUNBUFFERED=1 + +cd "${CURATOR_DIR}" +srun --ntasks=1 python -m tutorials.image.lance_writer.pipeline ingest \ + --manifest-dir "${MANIFEST_DIR}" \ + --dataset-uri "${DATASET_URI}" \ + --lance-commit-path "${LANCE_COMMIT_PATH}" \ + --checkpoint-path "${CHECKPOINT_PATH}" \ + --source-storage-options "${SOURCE_STORAGE_OPTIONS}" \ + --lance-storage-options "${LANCE_STORAGE_OPTIONS}" \ + --cpus "${CPUS}" \ + --ray-temp-dir "${RAY_TMPDIR}" diff --git a/uv.lock b/uv.lock index c45d948c1e..21864e54a7 100644 --- a/uv.lock +++ b/uv.lock @@ -5307,6 +5307,13 @@ image-cuda12 = [ { name = "torchvision", version = "0.25.0", source = { registry = "https://pypi.org/simple" }, marker = "(platform_machine != 'aarch64' and platform_machine != 'x86_64') or sys_platform != 'linux'" }, { name = "torchvision", version = "0.25.0+cu129", source = { registry = "https://download.pytorch.org/whl/cu129" }, marker = "(platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, ] +image-lance = [ + { name = "lance-ray" }, + { name = "pillow" }, + { name = "s3fs" }, + { name = "torchvision", version = "0.25.0", source = { registry = "https://pypi.org/simple" }, marker = "(platform_machine != 'aarch64' and platform_machine != 'x86_64') or sys_platform != 'linux'" }, + { name = "torchvision", version = "0.25.0+cu129", source = { registry = "https://download.pytorch.org/whl/cu129" }, marker = "(platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, +] inference-server = [ { name = "ai-dynamo", marker = "platform_machine == 'x86_64' and sys_platform != 'darwin'" }, { name = "boto3" }, @@ -5590,12 +5597,14 @@ requires-dist = [ { name = "nemo-curator", extras = ["deduplication-cuda12"], marker = "extra == 'math-cuda12'" }, { name = "nemo-curator", extras = ["deduplication-cuda12"], marker = "extra == 'text-cuda12'" }, { name = "nemo-curator", extras = ["image-cpu"], marker = "extra == 'image-cuda12'" }, + { name = "nemo-curator", extras = ["image-cpu"], marker = "extra == 'image-lance'" }, { name = "nemo-curator", extras = ["image-cuda12"], marker = "extra == 'all'" }, { name = "nemo-curator", extras = ["inference-server"], marker = "extra == 'all'" }, { name = "nemo-curator", extras = ["inference-server"], marker = "extra == 'sdg-cuda12'" }, { name = "nemo-curator", extras = ["interleaved-cpu"], marker = "extra == 'interleaved-cuda12'" }, { name = "nemo-curator", extras = ["interleaved-cuda12"], marker = "extra == 'all'" }, { name = "nemo-curator", extras = ["lance"], marker = "extra == 'all'" }, + { name = "nemo-curator", extras = ["lance"], marker = "extra == 'image-lance'" }, { name = "nemo-curator", extras = ["math-cpu"], marker = "extra == 'math-cuda12'" }, { name = "nemo-curator", extras = ["math-cuda12"], marker = "extra == 'all'" }, { name = "nemo-curator", extras = ["sdg-cpu"], marker = "extra == 'sdg-cuda12'" }, @@ -5654,6 +5663,7 @@ requires-dist = [ { name = "ray", extras = ["llm", "serve"], marker = "extra == 'inference-server'", specifier = ">=2.55.1" }, { name = "requests", marker = "extra == 'translation-nmt'" }, { name = "resiliparse", marker = "extra == 'text-cpu'" }, + { name = "s3fs", marker = "extra == 'image-lance'", specifier = ">=2024.12.0" }, { name = "s3fs", marker = "extra == 'interleaved-cpu'", specifier = ">=2024.12.0" }, { name = "s5cmd", marker = "extra == 'text-cpu'" }, { name = "sacrebleu", marker = "extra == 'translation-metrics'", specifier = ">=2.6.0" }, @@ -5693,7 +5703,7 @@ requires-dist = [ { name = "warcio", marker = "extra == 'text-cpu'" }, { name = "whisperx", marker = "platform_machine == 'x86_64' and sys_platform != 'darwin' and extra == 'audio-common'", specifier = ">=3.8.4" }, ] -provides-extras = ["cuda12", "vllm", "inference-server", "deduplication-cuda12", "audio-common", "audio-cpu", "audio-cuda12", "image-cpu", "image-cuda12", "translation-common", "translation-metrics", "translation-segmentation", "translation-aws", "translation-google", "translation-nmt", "translation-all", "text-cpu", "lance", "text-cuda12", "video-cpu", "video-cuda12", "math-cpu", "math-cuda12", "interleaved-cpu", "interleaved-cuda12", "sdg-cpu", "sdg-cuda12", "all"] +provides-extras = ["cuda12", "vllm", "inference-server", "deduplication-cuda12", "audio-common", "audio-cpu", "audio-cuda12", "image-cpu", "image-lance", "image-cuda12", "translation-common", "translation-metrics", "translation-segmentation", "translation-aws", "translation-google", "translation-nmt", "translation-all", "text-cpu", "lance", "text-cuda12", "video-cpu", "video-cuda12", "math-cpu", "math-cuda12", "interleaved-cpu", "interleaved-cuda12", "sdg-cpu", "sdg-cuda12", "all"] [package.metadata.requires-dev] build = [