cleverdatasets/dataset-uploader
0
0
Fork 0
Code Issues 31 Pull Requests 7 Actions 12 Packages Projects Releases Wiki Activity

fix: fix path too long issue; fix pickling issue in unified converter script #42

Merged
aditya merged 8 commits from path-too-long-fix into stream-download-convert-upload-merge-15-refactor-rdf-converters 2026-01-02 10:12:23 +00:00
Conversation 1 Commits 8 Files Changed 14
+557 -3447
14 changed files with 557 additions and 3447 deletions
Download Patch File Download Diff File Expand all files Collapse all files
features/steps/rdf_converter_steps.py
+10 -11
View File
@@ -14,23 +14,21 @@ import bz2
import gzip
import sys
from pathlib import Path
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING
from behave import given, then, when
from scripts.convert_rdf_to_hf_dataset_unified import (
ConversionConfig,
ConversionResult,
ProgressTracker,
StrategySelector,
)
# Add scripts directory to path for imports
SCRIPTS_DIR = Path(__file__).parent.parent.parent / "scripts"
sys.path.insert(0, str(SCRIPTS_DIR))
from convert_rdf_to_hf_dataset_unified import (
ConversionConfig,
ConversionResult,
FileHandler,
ProgressTracker,
SchemaManager,
StrategySelector,
)
if TYPE_CHECKING:
from behave.runner import Context
@@ -726,7 +724,8 @@ def step_verify_strategies_include(context: Context, strategy_names: str) -> Non
def _run_cli(args: list[str]) -> tuple[int, str, str]:
"""Run the CLI by calling main() directly for coverage tracking."""
import io
from contextlib import redirect_stdout, redirect_stderr
from contextlib import redirect_stderr, redirect_stdout
from convert_rdf_to_hf_dataset_unified import main
old_argv = sys.argv
pyproject.toml
+1 -1
View File
@@ -7,7 +7,7 @@ name = "app-boilerplate"
version = "0.1.0"
description = "Replace with a short summary of your project."
readme = "README.md"
requires-python = ">=3.9"
requires-python = ">=3.10"
license = {text = "MIT"}
authors = [
{name = "Your Name", email = "your.email@example.com"},
scripts/convert_rdf_to_hf_dataset.py
-1292
View File
File diff suppressed because it is too large Load Diff
scripts/convert_rdf_to_hf_dataset_streaming.py
-720
View File
@@ -1,720 +0,0 @@
#!/usr/bin/env python3
"""Convert RDF datasets to HuggingFace dataset format using streaming.
This minimizes memory usage.
This script parses RDF files in various formats (Turtle, N-Triples, RDF/XML, etc.)
and converts them to HuggingFace dataset format with lossless preservation of all
semantic information. Uses streaming and chunking to handle large datasets with
minimal memory footprint.
USAGE:
# Basic conversion
python scripts/convert_rdf_to_hf_dataset_streaming.py \\
<input_file> <output_dir> [options]
# Examples - Different RDF formats
python scripts/convert_rdf_to_hf_dataset_streaming.py \
raw_datasets/geonames/geonames.rdf \
hf_datasets/geonames \
--format xml \
--chunk-size 10000
ARGUMENTS:
input Path to input RDF file (any supported format)
output Path to output directory for HuggingFace dataset
-f, --format FORMAT RDF format: turtle, nt, ntriples, xml, n3, trig, nquads
(default: turtle)
--chunk-size SIZE Number of triples to process at once (default: 10000)
--max-workers NUM Maximum number of worker processes (default: CPU count)
--description TEXT Dataset description (added to metadata)
--citation TEXT Dataset citation (added to metadata)
--homepage URL Dataset homepage URL (added to metadata)
--license TEXT Dataset license (added to metadata)
-v, --verbose Show verbose output including library warnings
MEMORY OPTIMIZATIONS:
- Streams RDF file instead of loading entirely into memory
- Processes triples in configurable chunks
- Writes to Parquet files incrementally
- Uses generators to avoid materializing large lists
- Parallel processing for GeoNames format without memory accumulation
OUTPUT:
Creates HuggingFace dataset in Arrow/Parquet format with minimal memory usage.
"""
from __future__ import annotations
import argparse
import gzip
import json
import logging
import multiprocessing as mp
import shutil
import time
from collections.abc import Iterator
from pathlib import Path
from typing import Any, cast
import pyarrow as pa
import pyarrow.parquet as pq
from datasets import Dataset, DatasetDict
from rdflib import Graph, Literal, URIRef
from rich.console import Console
from rich.progress import (
BarColumn,
Progress,
SpinnerColumn,
TextColumn,
TimeElapsedColumn,
)
# Logger will be configured in main() based on --verbose flag
logger = logging.getLogger(__name__)
def _process_ntriples_file(
file_obj, chunk_size: int
) -> Iterator[list[dict[str, str]]]:
"""Process N-Triples file and yield chunks of triples.
This helper function processes the file object line by line, parsing each
line as an N-Triple and accumulating triples into chunks.
Args:
file_obj: File-like object (already opened)
chunk_size: Number of triples per chunk
Yields:
Chunks of triple dictionaries
"""
from rdflib import Graph
current_chunk = []
for line_no, line in enumerate(file_obj, 1):
line = line.strip()
# Skip empty lines and comments
if not line or line.startswith("#"):
continue
try:
# Parse single line as N-Triple
mini_graph = Graph()
mini_graph.parse(data=line, format="nt")
for s, p, o in mini_graph:
# Convert RDFLib objects to dictionary format
if isinstance(o, Literal):
object_type = "literal"
object_datatype = (
str(o.datatype) if o.datatype else None
)
object_language = o.language if o.language else None
elif isinstance(o, URIRef):
object_type = "uri"
object_datatype = None
object_language = None
else:
object_type = "blank_node"
object_datatype = None
object_language = None
triple = {
"subject": str(s),
"predicate": str(p),
"object": str(o),
"object_type": object_type,
"object_datatype": object_datatype,
"object_language": object_language,
}
current_chunk.append(triple)
# Yield chunk when it reaches the target size
if len(current_chunk) >= chunk_size:
yield current_chunk
current_chunk = []
except Exception as e:
# Skip malformed lines and continue processing
logger.debug(f"Skipping malformed line {line_no}: {e}")
continue
# Yield any remaining triples that didn't fill a complete chunk
if current_chunk:
yield current_chunk
def stream_ntriples(
file_path: Path, chunk_size: int = 10000
) -> Iterator[list[dict[str, str]]]:
"""Stream N-Triples file in chunks without loading entire file into memory.
Args:
file_path: Path to N-Triples file
chunk_size: Number of triples per chunk
Yields:
Chunks of triple dictionaries
"""
# Open file with appropriate context manager based on compression
# Context managers ensure files are properly closed even if exceptions occur
if file_path.suffix == ".gz":
with gzip.open(file_path, "rt", encoding="utf-8") as file_obj:
yield from _process_ntriples_file(file_obj, chunk_size)
else:
with open(file_path, encoding="utf-8") as file_obj:
yield from _process_ntriples_file(file_obj, chunk_size)
def stream_turtle_chunks(
file_path: Path, chunk_size: int = 10000
) -> Iterator[list[dict[str, str]]]:
"""Stream Turtle file in chunks using incremental parsing.
Args:
file_path: Path to Turtle file
chunk_size: Number of triples per chunk
Yields:
Chunks of triple dictionaries
"""
from rdflib import Graph
# For Turtle, we need to parse the entire file due to prefix definitions
# But we can yield results in chunks to avoid keeping all in memory
graph = Graph()
# Parse file
if file_path.suffix == ".gz":
with gzip.open(file_path, "rt", encoding="utf-8") as f:
graph.parse(f, format="turtle")
else:
graph.parse(str(file_path), format="turtle")
# Yield triples in chunks
current_chunk = []
for s, p, o in graph:
# Convert to dictionary format
if isinstance(o, Literal):
object_type = "literal"
object_datatype = str(o.datatype) if o.datatype else None
object_language = o.language if o.language else None
elif isinstance(o, URIRef):
object_type = "uri"
object_datatype = None
object_language = None
else:
object_type = "blank_node"
object_datatype = None
object_language = None
triple = {
"subject": str(s),
"predicate": str(p),
"object": str(o),
"object_type": object_type,
"object_datatype": object_datatype,
"object_language": object_language,
}
current_chunk.append(triple)
if len(current_chunk) >= chunk_size:
yield current_chunk
current_chunk = []
# Yield remaining triples
if current_chunk:
yield current_chunk
def process_geonames_segment(args):
"""Process a file segment for GeoNames parallel processing.
This function is defined at module level so it can be pickled for multiprocessing.
Args:
args: Tuple of (file_path, start_byte, end_byte, chunk_size)
Returns:
List of chunks of triple dictionaries
"""
from rdflib import Graph, Literal, URIRef
file_path, start_byte, end_byte, chunk_size = args
chunks = []
current_chunk = []
current_xml = []
with open(file_path, encoding="utf-8") as f:
f.seek(start_byte)
# Skip to next document boundary if not at start
if start_byte > 0:
while f.tell() < end_byte:
line = f.readline()
if not line:
break
if line.startswith("http://") or line.startswith("https://"):
break
# Process documents in segment
while f.tell() < end_byte:
line = f.readline()
if not line:
break
# Document boundary
if line.startswith("http://") or line.startswith("https://"):
if current_xml:
# Parse accumulated document
xml_str = "".join(current_xml)
try:
graph = Graph()
graph.parse(data=xml_str, format="xml")
for s, p, o in graph:
# Convert to dictionary format
if isinstance(o, Literal):
object_type = "literal"
object_datatype = (
str(o.datatype) if o.datatype else None
)
object_language = o.language if o.language else None
elif isinstance(o, URIRef):
object_type = "uri"
object_datatype = None
object_language = None
else:
object_type = "blank_node"
object_datatype = None
object_language = None
triple = {
"subject": str(s),
"predicate": str(p),
"object": str(o),
"object_type": object_type,
"object_datatype": object_datatype,
"object_language": object_language,
}
current_chunk.append(triple)
if len(current_chunk) >= chunk_size:
chunks.append(current_chunk)
current_chunk = []
except Exception:
pass
current_xml = []
continue
current_xml.append(line)
# Handle remaining data
if current_xml:
xml_str = "".join(current_xml)
try:
graph = Graph()
graph.parse(data=xml_str, format="xml")
for s, p, o in graph:
# Convert to dictionary format
if isinstance(o, Literal):
object_type = "literal"
object_datatype = str(o.datatype) if o.datatype else None
object_language = o.language if o.language else None
elif isinstance(o, URIRef):
object_type = "uri"
object_datatype = None
object_language = None
else:
object_type = "blank_node"
object_datatype = None
object_language = None
triple = {
"subject": str(s),
"predicate": str(p),
"object": str(o),
"object_type": object_type,
"object_datatype": object_datatype,
"object_language": object_language,
}
current_chunk.append(triple)
if len(current_chunk) >= chunk_size:
chunks.append(current_chunk)
current_chunk = []
except Exception:
pass
if current_chunk:
chunks.append(current_chunk)
return chunks
def stream_geonames_chunks(
file_path: Path, chunk_size: int = 10000, num_workers: int | None = None
) -> Iterator[list[dict[str, str]]]:
"""Stream GeoNames RDF file using parallel chunk processing.
Each worker processes a file segment independently and yields chunks.
Args:
file_path: Path to GeoNames RDF file
chunk_size: Number of triples per chunk
num_workers: Number of worker processes
Yields:
Chunks of triple dictionaries
"""
if num_workers is None:
num_workers = mp.cpu_count()
from multiprocessing import Pool
# Calculate file segments
file_size = file_path.stat().st_size
segment_size = file_size // num_workers
segments = []
with open(file_path, "rb") as f:
for i in range(num_workers):
start = i * segment_size
if i == num_workers - 1:
end = file_size
else:
end = start + segment_size
f.seek(end)
# Find next document boundary
while end < file_size:
line = f.readline()
end = f.tell()
decoded = line.decode("utf-8", errors="ignore")
if decoded.startswith("http://") or decoded.startswith("https://"):
break
segments.append((str(file_path), start, end, chunk_size))
# Process segments in parallel using the module-level function
with Pool(processes=num_workers) as pool:
results = pool.map(process_geonames_segment, segments)
# Yield chunks from all workers
for worker_chunks in results:
yield from worker_chunks
def stream_rdf_chunks(
file_path: Path,
format: str = "turtle",
chunk_size: int = 10000,
num_workers: int | None = None,
) -> Iterator[list[dict[str, str]]]:
"""Stream RDF file in chunks based on format.
Args:
file_path: Path to RDF file
format: RDF format
chunk_size: Number of triples per chunk
num_workers: Number of workers for parallel formats
Yields:
Chunks of triple dictionaries
"""
console = Console()
# Check if this is GeoNames format
is_geonames = "geonames" in str(file_path).lower()
if is_geonames and format in ("xml", "application/rdf+xml"):
with open(file_path, encoding="utf-8") as f:
first_line = f.readline().strip()
if first_line.startswith("http://") or first_line.startswith("https://"):
console.print("[yellow]Using GeoNames streaming parser[/yellow]")
yield from stream_geonames_chunks(file_path, chunk_size, num_workers)
return
# Use appropriate streaming method based on format
if format in ("nt", "ntriples"):
console.print("[yellow]Using N-Triples streaming parser[/yellow]")
yield from stream_ntriples(file_path, chunk_size)
elif format == "turtle":
console.print("[yellow]Using Turtle chunk parser[/yellow]")
yield from stream_turtle_chunks(file_path, chunk_size)
else:
# For other formats, fall back to regular parsing with chunked output
console.print(f"[yellow]Using generic RDF parser for {format}[/yellow]")
graph = Graph()
graph.parse(str(file_path), format=format)
current_chunk = []
for s, p, o in graph:
# Convert to dictionary format
if isinstance(o, Literal):
object_type = "literal"
object_datatype = str(o.datatype) if o.datatype else None
object_language = o.language if o.language else None
elif isinstance(o, URIRef):
object_type = "uri"
object_datatype = None
object_language = None
else:
object_type = "blank_node"
object_datatype = None
object_language = None
triple = {
"subject": str(s),
"predicate": str(p),
"object": str(o),
"object_type": object_type,
"object_datatype": object_datatype,
"object_language": object_language,
}
current_chunk.append(triple)
if len(current_chunk) >= chunk_size:
yield current_chunk
current_chunk = []
if current_chunk:
yield current_chunk
def convert_rdf_to_hf_streaming(
input_path: Path,
output_path: Path,
rdf_format: str = "turtle",
chunk_size: int = 10000,
num_workers: int | None = None,
metadata: dict[str, Any] | None = None,
) -> None:
"""Convert an RDF file to HuggingFace dataset format using streaming.
Args:
input_path: Path to input RDF file
output_path: Path to output directory
rdf_format: RDF format (turtle, nt, xml, etc.)
chunk_size: Number of triples to process at once
num_workers: Number of worker processes for parallel formats
metadata: Optional metadata dictionary
"""
console = Console()
# Create output directory
output_path.mkdir(parents=True, exist_ok=True)
# Get file size for progress tracking
file_size_mb = input_path.stat().st_size / (1024 * 1024)
console.print(f"[cyan]Converting RDF file: {input_path}[/cyan]")
console.print(f"[dim]File size: {file_size_mb:.2f} MB ({rdf_format} format)[/dim]")
console.print(f"[dim]Chunk size: {chunk_size:,} triples[/dim]")
console.print(f"[dim]Output: {output_path}[/dim]\n")
# Define schema
schema = pa.schema(
[
("subject", pa.string()),
("predicate", pa.string()),
("object", pa.string()),
("object_type", pa.string()),
("object_datatype", pa.string()),
("object_language", pa.string()),
]
)
# Process and write chunks
data_dir = output_path / "data"
data_dir.mkdir(exist_ok=True)
total_triples = 0
chunk_count = 0
parquet_files = []
with Progress(
SpinnerColumn(),
TextColumn("[progress.description]{task.description}"),
BarColumn(),
TextColumn("{task.completed:,} chunks"),
TextColumn("•"),
TextColumn("{task.fields[triples]:,} triples"),
TimeElapsedColumn(),
console=console,
transient=False,
) as progress:
task = progress.add_task(
"[yellow]Processing RDF chunks...[/yellow]",
total=None,
triples=0,
)
# Stream and process chunks
for chunk in stream_rdf_chunks(input_path, rdf_format, chunk_size, num_workers):
# Convert chunk to Arrow table
table = pa.Table.from_pylist(chunk, schema=schema)
# Write chunk to Parquet file
chunk_file = data_dir / f"chunk-{chunk_count:05d}.parquet"
pq.write_table(table, chunk_file)
parquet_files.append(chunk_file)
# Update progress
total_triples += len(chunk)
chunk_count += 1
progress.update(
task,
advance=1,
completed=chunk_count,
triples=total_triples,
)
console.print(
f"\n[green]✓ Processed {total_triples:,} triples in "
f"{chunk_count} chunks[/green]"
)
# Merge Parquet files into final dataset structure
console.print("\n[yellow]Merging chunks into final dataset...[/yellow]")
# Read all Parquet files as a single dataset
# Note: from_parquet can return different types, but with a glob pattern
# it returns a Dataset that supports train_test_split
dataset = cast(Dataset, Dataset.from_parquet(str(data_dir / "*.parquet")))
# Split into train/test
console.print("[dim]Creating train/test split (95%/5%)...[/dim]")
train_test = dataset.train_test_split(test_size=0.05, seed=42)
# Create DatasetDict
dataset_dict = DatasetDict(
{
"train": train_test["train"],
"test": train_test["test"],
}
)
# Add metadata
if metadata:
for split in dataset_dict:
dataset_dict[split].info.description = metadata.get("description", "")
dataset_dict[split].info.citation = metadata.get("citation", "")
dataset_dict[split].info.homepage = metadata.get("homepage", "")
dataset_dict[split].info.license = metadata.get("license", "")
# Save final dataset
console.print("[yellow]Saving final dataset...[/yellow]")
dataset_dict.save_to_disk(str(output_path))
# Clean up temporary chunk files
console.print("[dim]Cleaning up temporary files...[/dim]")
shutil.rmtree(data_dir)
# Save dataset info
info = {
"format": "parquet",
"total_triples": total_triples,
"train_size": len(dataset_dict["train"]),
"test_size": len(dataset_dict["test"]),
"source_format": rdf_format,
"chunk_size": chunk_size,
"conversion_date": time.strftime("%Y-%m-%d %H:%M:%S"),
}
with open(output_path / "dataset_info.json", "w") as f:
json.dump(info, f, indent=2)
console.print(
"\n[bold green]✓ Successfully converted to HuggingFace dataset[/bold green]"
)
console.print(f"[green] Train: {len(dataset_dict['train']):,} triples[/green]")
console.print(f"[green] Test: {len(dataset_dict['test']):,} triples[/green]")
console.print(f"[green] Location: {output_path}[/green]")
def main():
"""Main entry point."""
parser = argparse.ArgumentParser(
description="Convert RDF to HuggingFace dataset format with streaming",
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument("input", type=Path, help="Input RDF file")
parser.add_argument(
"output", type=Path, help="Output directory for HuggingFace dataset"
)
parser.add_argument(
"-f",
"--format",
default="turtle",
choices=["turtle", "nt", "ntriples", "xml", "n3", "trig", "nquads"],
help="RDF format (default: turtle)",
)
parser.add_argument(
"--chunk-size",
type=int,
default=10000,
help="Number of triples to process at once (default: 10000)",
)
parser.add_argument(
"--max-workers",
type=int,
default=None,
help="Maximum number of worker processes (default: CPU count)",
)
parser.add_argument("--description", help="Dataset description")
parser.add_argument("--citation", help="Dataset citation")
parser.add_argument("--homepage", help="Dataset homepage URL")
parser.add_argument("--license", help="Dataset license")
parser.add_argument(
"-v", "--verbose", action="store_true", help="Show verbose output"
)
args = parser.parse_args()
# Configure logging
if args.verbose:
logging.basicConfig(level=logging.INFO)
else:
logging.basicConfig(level=logging.ERROR)
# Validate input
if not args.input.exists():
print(f"Error: Input file not found: {args.input}")
return 1
# Prepare metadata
metadata = {}
if args.description:
metadata["description"] = args.description
if args.citation:
metadata["citation"] = args.citation
if args.homepage:
metadata["homepage"] = args.homepage
if args.license:
metadata["license"] = args.license
# Convert
try:
convert_rdf_to_hf_streaming(
input_path=args.input,
output_path=args.output,
rdf_format=args.format,
chunk_size=args.chunk_size,
num_workers=args.max_workers,
metadata=metadata,
)
return 0
except Exception as e:
console = Console()
console.print(f"[red]Error: {e}[/red]")
if args.verbose:
import traceback
traceback.print_exc()
return 1
if __name__ == "__main__":
exit(main())
scripts/convert_rdf_to_hf_dataset_streaming_parallel.py
+8 -13
View File
@@ -11,7 +11,6 @@ import gzip
import json
import logging
import multiprocessing as mp
import shutil
import sys
import time
from collections.abc import Iterator
@@ -19,17 +18,9 @@ from pathlib import Path
from typing import Any
import pyarrow as pa
import pyarrow.parquet as pq
from datasets import Dataset, DatasetDict, Features, Value
from datasets import Dataset, DatasetDict, Features, IterableDataset, Value
from rdflib import Graph, Literal, URIRef
from rich.console import Console
from rich.progress import (
BarColumn,
Progress,
SpinnerColumn,
TextColumn,
TimeElapsedColumn,
)
logger = logging.getLogger(__name__)
logger.addHandler(logging.StreamHandler(sys.stdout))
@@ -655,7 +646,7 @@ def convert_rdf_to_hf_streaming(
# unpicklable objects. Dataset.from_generator() requires picklable
# generators, so using simple print statements
def dataset_generator():
triple_count = 0
nonlocal triple_count
chunk_count = 0
last_print_count = 0
@@ -688,16 +679,20 @@ def convert_rdf_to_hf_streaming(
'object_language': Value('string'),
})
triple_count = 0
dataset = Dataset.from_generator(dataset_generator, features=features)
print("\nPROGRESS: 75", flush=True)
console.print("[green]✓ Dataset generation complete[/green]")
# Get statistics after generation
total_triples = len(dataset)
if not isinstance(dataset, IterableDataset):
total_triples = len(dataset)
else:
total_triples = triple_count
# Wrap in DatasetDict
dataset_dict = DatasetDict({"data": dataset})
dataset_dict = DatasetDict({"data": dataset}) # pyright: ignore
# Add metadata
if metadata:
scripts/convert_rdf_to_hf_dataset_streaming_simple.py
-672
View File
@@ -1,672 +0,0 @@
#!/usr/bin/env python3
"""Simple streaming converter for RDF to HuggingFace dataset format with
minimal memory usage.
This version uses a simpler single-threaded approach that's more reliable and still
memory-efficient.
"""
from __future__ import annotations
import argparse
import gzip
import json
import logging
import shutil
import time
from collections.abc import Iterator
from pathlib import Path
from typing import Any
import pyarrow as pa
import pyarrow.parquet as pq
from datasets import Dataset, DatasetDict
from rdflib import Graph, Literal, URIRef
from rich.console import Console
from rich.progress import (
BarColumn,
Progress,
SpinnerColumn,
TextColumn,
TimeElapsedColumn,
)
logger = logging.getLogger(__name__)
def stream_geonames_simple(
file_path: Path, chunk_size: int = 10000
) -> Iterator[list[dict[str, str]]]:
"""Stream GeoNames RDF file with simple line-by-line processing.
Args:
file_path: Path to GeoNames RDF file
chunk_size: Number of triples per chunk
Yields:
Chunks of triple dictionaries
"""
current_chunk = []
current_xml = []
doc_count = 0
# Handle compressed files
if file_path.suffix == ".gz":
with gzip.open(file_path, "rt", encoding="utf-8") as file_obj:
for _line_no, line in enumerate(file_obj, 1):
# Check if this is a document boundary (URL line)
if line.startswith("http://") or line.startswith("https://"):
# Process previous document if exists
if current_xml:
xml_str = "".join(current_xml)
try:
# Parse the RDF/XML document
graph = Graph()
graph.parse(data=xml_str, format="xml")
# Extract triples
for s, p, o in graph:
# Convert to dictionary format
if isinstance(o, Literal):
object_type = "literal"
object_datatype = (
str(o.datatype) if o.datatype else None
)
object_language = o.language if o.language else None
elif isinstance(o, URIRef):
object_type = "uri"
object_datatype = None
object_language = None
else:
object_type = "blank_node"
object_datatype = None
object_language = None
triple = {
"subject": str(s),
"predicate": str(p),
"object": str(o),
"object_type": object_type,
"object_datatype": object_datatype,
"object_language": object_language,
}
current_chunk.append(triple)
# Yield chunk if it's full
if len(current_chunk) >= chunk_size:
yield current_chunk
current_chunk = []
doc_count += 1
if doc_count % 1000 == 0:
logger.info(f"Processed {doc_count} documents")
except Exception as e:
# Skip malformed documents
logger.debug(f"Error parsing document {doc_count}: {e}")
# Reset for next document
current_xml = []
# Skip the URL line itself
continue
# Accumulate XML content
current_xml.append(line)
# Process final document if exists
if current_xml:
xml_str = "".join(current_xml)
try:
graph = Graph()
graph.parse(data=xml_str, format="xml")
for s, p, o in graph:
if isinstance(o, Literal):
object_type = "literal"
object_datatype = str(o.datatype) if o.datatype else None
object_language = o.language if o.language else None
elif isinstance(o, URIRef):
object_type = "uri"
object_datatype = None
object_language = None
else:
object_type = "blank_node"
object_datatype = None
object_language = None
triple = {
"subject": str(s),
"predicate": str(p),
"object": str(o),
"object_type": object_type,
"object_datatype": object_datatype,
"object_language": object_language,
}
current_chunk.append(triple)
if len(current_chunk) >= chunk_size:
yield current_chunk
current_chunk = []
except Exception as e:
logger.debug(f"Error parsing final document: {e}")
# Yield remaining triples
if current_chunk:
yield current_chunk
else:
with open(file_path, encoding="utf-8") as file_obj:
for _line_no, line in enumerate(file_obj, 1):
# Check if this is a document boundary (URL line)
if line.startswith("http://") or line.startswith("https://"):
# Process previous document if exists
if current_xml:
xml_str = "".join(current_xml)
try:
# Parse the RDF/XML document
graph = Graph()
graph.parse(data=xml_str, format="xml")
# Extract triples
for s, p, o in graph:
# Convert to dictionary format
if isinstance(o, Literal):
object_type = "literal"
object_datatype = (
str(o.datatype) if o.datatype else None
)
object_language = (
o.language if o.language else None
)
elif isinstance(o, URIRef):
object_type = "uri"
object_datatype = None
object_language = None
else:
object_type = "blank_node"
object_datatype = None
object_language = None
triple = {
"subject": str(s),
"predicate": str(p),
"object": str(o),
"object_type": object_type,
"object_datatype": object_datatype,
"object_language": object_language,
}
current_chunk.append(triple)
# Yield chunk if it's full
if len(current_chunk) >= chunk_size:
yield current_chunk
current_chunk = []
doc_count += 1
if doc_count % 1000 == 0:
logger.info(f"Processed {doc_count} documents")
except Exception as e:
# Skip malformed documents
logger.debug(f"Error parsing document {doc_count}: {e}")
# Reset for next document
current_xml = []
# Skip the URL line itself
continue
# Accumulate XML content
current_xml.append(line)
# Process final document if exists
if current_xml:
xml_str = "".join(current_xml)
try:
graph = Graph()
graph.parse(data=xml_str, format="xml")
for s, p, o in graph:
if isinstance(o, Literal):
object_type = "literal"
object_datatype = str(o.datatype) if o.datatype else None
object_language = o.language if o.language else None
elif isinstance(o, URIRef):
object_type = "uri"
object_datatype = None
object_language = None
else:
object_type = "blank_node"
object_datatype = None
object_language = None
triple = {
"subject": str(s),
"predicate": str(p),
"object": str(o),
"object_type": object_type,
"object_datatype": object_datatype,
"object_language": object_language,
}
current_chunk.append(triple)
if len(current_chunk) >= chunk_size:
yield current_chunk
current_chunk = []
except Exception as e:
logger.debug(f"Error parsing final document: {e}")
# Yield remaining triples
if current_chunk:
yield current_chunk
def stream_ntriples_simple(
file_path: Path, chunk_size: int = 10000
) -> Iterator[list[dict[str, str]]]:
"""Stream N-Triples file line by line.
Args:
file_path: Path to N-Triples file
chunk_size: Number of triples per chunk
Yields:
Chunks of triple dictionaries
"""
current_chunk = []
# Handle compressed files
if file_path.suffix == ".gz":
with gzip.open(file_path, "rt", encoding="utf-8") as file_obj:
# Create a mini graph for parsing individual lines
for _line_no, line in enumerate(file_obj, 1):
line = line.strip()
if not line or line.startswith("#"):
continue
try:
# Parse single N-Triple line
mini_graph = Graph()
mini_graph.parse(data=line, format="nt")
for s, p, o in mini_graph:
# Convert to dictionary format
if isinstance(o, Literal):
object_type = "literal"
object_datatype = str(o.datatype) if o.datatype else None
object_language = o.language if o.language else None
elif isinstance(o, URIRef):
object_type = "uri"
object_datatype = None
object_language = None
else:
object_type = "blank_node"
object_datatype = None
object_language = None
triple = {
"subject": str(s),
"predicate": str(p),
"object": str(o),
"object_type": object_type,
"object_datatype": object_datatype,
"object_language": object_language,
}
current_chunk.append(triple)
if len(current_chunk) >= chunk_size:
yield current_chunk
current_chunk = []
except Exception as e:
# Skip malformed lines
logger.debug(f"Skipping malformed line {_line_no}: {e}")
continue
# Yield remaining triples
if current_chunk:
yield current_chunk
else:
with open(file_path, encoding="utf-8") as file_obj:
# Create a mini graph for parsing individual lines
for _line_no, line in enumerate(file_obj, 1):
line = line.strip()
if not line or line.startswith("#"):
continue
try:
# Parse single N-Triple line
mini_graph = Graph()
mini_graph.parse(data=line, format="nt")
for s, p, o in mini_graph:
# Convert to dictionary format
if isinstance(o, Literal):
object_type = "literal"
object_datatype = str(o.datatype) if o.datatype else None
object_language = o.language if o.language else None
elif isinstance(o, URIRef):
object_type = "uri"
object_datatype = None
object_language = None
else:
object_type = "blank_node"
object_datatype = None
object_language = None
triple = {
"subject": str(s),
"predicate": str(p),
"object": str(o),
"object_type": object_type,
"object_datatype": object_datatype,
"object_language": object_language,
}
current_chunk.append(triple)
if len(current_chunk) >= chunk_size:
yield current_chunk
current_chunk = []
except Exception as e:
# Skip malformed lines
logger.debug(f"Skipping malformed line {_line_no}: {e}")
continue
# Yield remaining triples
if current_chunk:
yield current_chunk
def stream_rdf_chunks(
file_path: Path, format: str = "turtle", chunk_size: int = 10000
) -> Iterator[list[dict[str, str]]]:
"""Stream RDF file in chunks based on format.
Args:
file_path: Path to RDF file
format: RDF format
chunk_size: Number of triples per chunk
Yields:
Chunks of triple dictionaries
"""
console = Console()
# Check if this is GeoNames format
is_geonames = "geonames" in str(file_path).lower()
if is_geonames and format in ("xml", "application/rdf+xml"):
with open(file_path, encoding="utf-8", errors="ignore") as f:
first_line = f.readline().strip()
if first_line.startswith("http://") or first_line.startswith("https://"):
console.print(
"[yellow]Using GeoNames streaming parser (single-threaded)[/yellow]"
)
yield from stream_geonames_simple(file_path, chunk_size)
return
# Use appropriate streaming method based on format
if format in ("nt", "ntriples"):
console.print("[yellow]Using N-Triples streaming parser[/yellow]")
yield from stream_ntriples_simple(file_path, chunk_size)
else:
# For other formats, we need to parse the whole file but yield in chunks
console.print(f"[yellow]Using standard RDF parser for {format}[/yellow]")
console.print(
"[dim]Note: Non-streaming formats require loading the full graph[/dim]"
)
graph = Graph()
graph.parse(str(file_path), format=format)
current_chunk = []
for s, p, o in graph:
# Convert to dictionary format
if isinstance(o, Literal):
object_type = "literal"
object_datatype = str(o.datatype) if o.datatype else None
object_language = o.language if o.language else None
elif isinstance(o, URIRef):
object_type = "uri"
object_datatype = None
object_language = None
else:
object_type = "blank_node"
object_datatype = None
object_language = None
triple = {
"subject": str(s),
"predicate": str(p),
"object": str(o),
"object_type": object_type,
"object_datatype": object_datatype,
"object_language": object_language,
}
current_chunk.append(triple)
if len(current_chunk) >= chunk_size:
yield current_chunk
current_chunk = []
if current_chunk:
yield current_chunk
def convert_rdf_to_hf_streaming(
input_path: Path,
output_path: Path,
rdf_format: str = "turtle",
chunk_size: int = 10000,
metadata: dict[str, Any] | None = None,
) -> None:
"""Convert an RDF file to HuggingFace dataset format using streaming.
Args:
input_path: Path to input RDF file
output_path: Path to output directory
rdf_format: RDF format (turtle, nt, xml, etc.)
chunk_size: Number of triples to process at once
metadata: Optional metadata dictionary
"""
console = Console()
# Create output directory
output_path.mkdir(parents=True, exist_ok=True)
# Get file size for progress tracking
file_size_mb = input_path.stat().st_size / (1024 * 1024)
console.print(f"[cyan]Converting RDF file: {input_path.name}[/cyan]")
console.print(f"[dim]File size: {file_size_mb:.2f} MB ({rdf_format} format)[/dim]")
console.print(f"[dim]Chunk size: {chunk_size:,} triples[/dim]")
console.print(f"[dim]Output: {output_path}[/dim]\n")
# Define schema
schema = pa.schema(
[
("subject", pa.string()),
("predicate", pa.string()),
("object", pa.string()),
("object_type", pa.string()),
("object_datatype", pa.string()),
("object_language", pa.string()),
]
)
# Process and write chunks
temp_chunks_dir = output_path / "temp_chunks"
temp_chunks_dir.mkdir(exist_ok=True)
total_triples = 0
chunk_count = 0
parquet_files = []
with Progress(
SpinnerColumn(),
TextColumn("[progress.description]{task.description}"),
BarColumn(),
TextColumn("{task.completed:,} chunks"),
TextColumn("•"),
TextColumn("{task.fields[triples]:,} triples"),
TimeElapsedColumn(),
console=console,
transient=False,
) as progress:
task = progress.add_task(
"[yellow]Processing RDF chunks...[/yellow]",
total=None,
triples=0,
)
# Stream and process chunks
for chunk in stream_rdf_chunks(input_path, rdf_format, chunk_size):
# Convert chunk to Arrow table
table = pa.Table.from_pylist(chunk, schema=schema)
# Write chunk to Parquet file
chunk_file = temp_chunks_dir / f"chunk-{chunk_count:05d}.parquet"
pq.write_table(table, chunk_file)
parquet_files.append(chunk_file)
# Update progress
total_triples += len(chunk)
chunk_count += 1
progress.update(
task,
advance=1,
completed=chunk_count,
triples=total_triples,
description=(
f"[yellow]Processing chunks ({file_size_mb:.1f} MB "
f"file)...[/yellow]"
),
)
console.print(
f"\n[green]✓ Processed {total_triples:,} triples in "
f"{chunk_count} chunks[/green]"
)
# Merge Parquet files into final dataset structure
console.print("\n[yellow]Merging chunks into final dataset...[/yellow]")
# Read all Parquet files as a single dataset
dataset = Dataset.from_parquet(str(temp_chunks_dir / "*.parquet"))
# Create DatasetDict with single 'data' split
dataset_dict = DatasetDict({"data": dataset})
# Add metadata
if metadata:
for split in dataset_dict:
dataset_dict[split].info.description = metadata.get("description", "")
dataset_dict[split].info.citation = metadata.get("citation", "")
dataset_dict[split].info.homepage = metadata.get("homepage", "")
dataset_dict[split].info.license = metadata.get("license", "")
# Save final dataset
console.print("[yellow]Saving final dataset...[/yellow]")
dataset_dict.save_to_disk(str(output_path))
# Clean up temporary chunk files
console.print("[dim]Cleaning up temporary files...[/dim]")
shutil.rmtree(temp_chunks_dir)
# Save dataset info
info = {
"format": "parquet",
"total_triples": total_triples,
"data_size": len(dataset_dict["data"]),
"source_format": rdf_format,
"chunk_size": chunk_size,
"conversion_date": time.strftime("%Y-%m-%d %H:%M:%S"),
}
with open(output_path / "dataset_info.json", "w") as f:
json.dump(info, f, indent=2)
console.print(
"\n[bold green]✓ Successfully converted to HuggingFace dataset[/bold green]"
)
console.print(f"[green] Data split: {len(dataset_dict['data']):,} triples[/green]")
console.print(f"[green] Location: {output_path}[/green]")
def main():
"""Main entry point."""
parser = argparse.ArgumentParser(
description="Convert RDF to HuggingFace dataset format with streaming",
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument("input", type=Path, help="Input RDF file")
parser.add_argument(
"output", type=Path, help="Output directory for HuggingFace dataset"
)
parser.add_argument(
"-f",
"--format",
default="turtle",
choices=["turtle", "nt", "ntriples", "xml", "n3", "trig", "nquads"],
help="RDF format (default: turtle)",
)
parser.add_argument(
"--chunk-size",
type=int,
default=10000,
help="Number of triples to process at once (default: 10000)",
)
parser.add_argument("--description", help="Dataset description")
parser.add_argument("--citation", help="Dataset citation")
parser.add_argument("--homepage", help="Dataset homepage URL")
parser.add_argument("--license", help="Dataset license")
parser.add_argument(
"-v", "--verbose", action="store_true", help="Show verbose output"
)
args = parser.parse_args()
# Configure logging
if args.verbose:
logging.basicConfig(level=logging.INFO)
else:
logging.basicConfig(level=logging.ERROR)
# Validate input
if not args.input.exists():
print(f"Error: Input file not found: {args.input}")
return 1
# Prepare metadata
metadata = {}
if args.description:
metadata["description"] = args.description
if args.citation:
metadata["citation"] = args.citation
if args.homepage:
metadata["homepage"] = args.homepage
if args.license:
metadata["license"] = args.license
# Convert
try:
convert_rdf_to_hf_streaming(
input_path=args.input,
output_path=args.output,
rdf_format=args.format,
chunk_size=args.chunk_size,
metadata=metadata,
)
return 0
except Exception as e:
console = Console()
console.print(f"[red]Error: {e}[/red]")
if args.verbose:
import traceback
traceback.print_exc()
return 1
if __name__ == "__main__":
exit(main())
scripts/convert_rdf_to_hf_dataset_streaming_turtle.py
-496
View File
@@ -1,496 +0,0 @@
#!/usr/bin/env python3
"""Streaming converter for RDF Turtle files to HuggingFace dataset format.
This version implements proper streaming for Turtle files by parsing them in chunks.
Turtle format requires more careful parsing than N-Triples due to prefixes
and multi-line statements.
"""
from __future__ import annotations
import argparse
import gzip
import json
import logging
import re
import shutil
import time
from collections.abc import Iterator
from pathlib import Path
from typing import Any, cast
import pyarrow as pa
import pyarrow.parquet as pq
from datasets import Dataset, DatasetDict
from rdflib import Graph, Literal, URIRef
from rich.console import Console
from rich.progress import (
BarColumn,
Progress,
SpinnerColumn,
TextColumn,
TimeElapsedColumn,
)
logger = logging.getLogger(__name__)
def parse_turtle_chunk(
chunk_text: str, prefixes: dict[str, str]
) -> list[dict[str, str]]:
"""Parse a chunk of Turtle text with given prefixes.
Args:
chunk_text: Turtle text chunk to parse
prefixes: Dictionary of prefix mappings
Returns:
List of triple dictionaries
"""
triples = []
# Build complete Turtle document with prefixes
doc_lines = []
for prefix, uri in prefixes.items():
doc_lines.append(f"@prefix {prefix}: <{uri}> .")
doc_lines.append("")
doc_lines.append(chunk_text)
try:
# Parse the chunk
graph = Graph()
graph.parse(data="\n".join(doc_lines), format="turtle")
# Extract triples
for s, p, o in graph:
if isinstance(o, Literal):
object_type = "literal"
object_datatype = str(o.datatype) if o.datatype else None
object_language = o.language if o.language else None
elif isinstance(o, URIRef):
object_type = "uri"
object_datatype = None
object_language = None
else:
object_type = "blank_node"
object_datatype = None
object_language = None
triple = {
"subject": str(s),
"predicate": str(p),
"object": str(o),
"object_type": object_type,
"object_datatype": object_datatype,
"object_language": object_language,
}
triples.append(triple)
except Exception as e:
# Log parse errors but continue
logger.debug(f"Error parsing chunk: {e}")
return triples
def stream_turtle_file(
file_path: Path, chunk_size: int = 10000
) -> Iterator[list[dict[str, str]]]:
"""Stream Turtle file in chunks, handling prefixes and multi-line statements.
Args:
file_path: Path to Turtle file
chunk_size: Approximate number of statements per chunk
Yields:
Chunks of triple dictionaries
"""
console = Console()
console.print(f"[yellow]Streaming Turtle file: {file_path.name}[/yellow]")
# Detect if file is gzipped
is_gzipped = file_path.suffix == ".gz"
if is_gzipped:
with gzip.open(
file_path, "rt", encoding="utf-8", errors="ignore"
) as file_obj:
prefixes = {}
current_chunk = []
statement_count = 0
line_count = 0
in_multiline = False
multiline_buffer = []
for line in file_obj:
line_count += 1
line = line.strip()
# Skip empty lines and comments
if not line or line.startswith("#"):
continue
# Handle prefix declarations
if line.startswith("@prefix"):
# Extract prefix and URI
match = re.match(r"@prefix\s+(\w+):\s*<([^>]+)>", line)
if match:
prefixes[match.group(1)] = match.group(2)
continue
elif line.startswith("@base"):
# Handle base declarations
continue
# Handle multi-line statements (simple heuristic)
if in_multiline:
multiline_buffer.append(line)
if line.endswith("."):
# End of statement
complete_statement = " ".join(multiline_buffer)
current_chunk.append(complete_statement)
statement_count += 1
in_multiline = False
multiline_buffer = []
else:
# Check if this is start of a multi-line statement
if (
not line.endswith(".")
and not line.endswith(";")
and not line.endswith(",")
):
# Start of multi-line
in_multiline = True
multiline_buffer = [line]
else:
# Complete single-line statement
current_chunk.append(line)
statement_count += 1
# Process chunk when it reaches target size
if statement_count >= chunk_size:
# Parse and yield the chunk
chunk_text = "\n".join(current_chunk)
triples = parse_turtle_chunk(chunk_text, prefixes)
if triples:
yield triples
# Reset for next chunk
current_chunk = []
statement_count = 0
# Log progress periodically
if line_count % 100000 == 0:
console.print(f"[dim]Processed {line_count:,} lines...[/dim]")
# Process remaining statements
if current_chunk:
chunk_text = "\n".join(current_chunk)
triples = parse_turtle_chunk(chunk_text, prefixes)
if triples:
yield triples
else:
with open(file_path, encoding="utf-8", errors="ignore") as file_obj:
prefixes = {}
current_chunk = []
statement_count = 0
line_count = 0
in_multiline = False
multiline_buffer = []
for line in file_obj:
line_count += 1
line = line.strip()
# Skip empty lines and comments
if not line or line.startswith("#"):
continue
# Handle prefix declarations
if line.startswith("@prefix"):
# Extract prefix and URI
match = re.match(r"@prefix\s+(\w+):\s*<([^>]+)>", line)
if match:
prefixes[match.group(1)] = match.group(2)
continue
elif line.startswith("@base"):
# Handle base declarations
continue
# Handle multi-line statements (simple heuristic)
if in_multiline:
multiline_buffer.append(line)
if line.endswith("."):
# End of statement
complete_statement = " ".join(multiline_buffer)
current_chunk.append(complete_statement)
statement_count += 1
in_multiline = False
multiline_buffer = []
else:
# Check if this is start of a multi-line statement
if (
not line.endswith(".")
and not line.endswith(";")
and not line.endswith(",")
):
# Start of multi-line
in_multiline = True
multiline_buffer = [line]
else:
# Complete single-line statement
current_chunk.append(line)
statement_count += 1
# Process chunk when it reaches target size
if statement_count >= chunk_size:
# Parse and yield the chunk
chunk_text = "\n".join(current_chunk)
triples = parse_turtle_chunk(chunk_text, prefixes)
if triples:
yield triples
# Reset for next chunk
current_chunk = []
statement_count = 0
# Log progress periodically
if line_count % 100000 == 0:
console.print(f"[dim]Processed {line_count:,} lines...[/dim]")
# Process remaining statements
if current_chunk:
chunk_text = "\n".join(current_chunk)
triples = parse_turtle_chunk(chunk_text, prefixes)
if triples:
yield triples
def process_turtle_chunk_parallel(args):
"""Process a Turtle chunk in parallel (for future optimization).
Module-level function for multiprocessing.
"""
chunk_text, prefixes = args
return parse_turtle_chunk(chunk_text, prefixes)
def convert_turtle_to_hf_streaming(
input_path: Path,
output_path: Path,
chunk_size: int = 10000,
metadata: dict[str, Any] | None = None,
) -> None:
"""Convert a Turtle RDF file to HuggingFace dataset format using streaming.
Args:
input_path: Path to input Turtle file
output_path: Path to output directory
chunk_size: Number of statements to process at once
metadata: Optional metadata dictionary
"""
console = Console()
# Create output directory
output_path.mkdir(parents=True, exist_ok=True)
# Get file size for progress tracking
file_size_mb = input_path.stat().st_size / (1024 * 1024)
console.print(f"[cyan]Converting Turtle RDF file: {input_path.name}[/cyan]")
console.print(f"[dim]File size: {file_size_mb:.2f} MB[/dim]")
console.print(f"[dim]Chunk size: {chunk_size:,} statements per batch[/dim]")
console.print(f"[dim]Output: {output_path}[/dim]\n")
# Emit initial progress
print("\nPROGRESS: 5", flush=True)
# Define schema
schema = pa.schema(
[
("subject", pa.string()),
("predicate", pa.string()),
("object", pa.string()),
("object_type", pa.string()),
("object_datatype", pa.string()),
("object_language", pa.string()),
]
)
# Process and write chunks
temp_chunks_dir = output_path / "temp_chunks"
temp_chunks_dir.mkdir(exist_ok=True)
total_triples = 0
chunk_count = 0
parquet_files = []
with Progress(
SpinnerColumn(),
TextColumn("[progress.description]{task.description}"),
BarColumn(),
TextColumn("{task.completed:,} chunks"),
TextColumn("•"),
TextColumn("{task.fields[triples]:,} triples"),
TextColumn("•"),
TextColumn("{task.fields[rate]:.0f} triples/sec"),
TimeElapsedColumn(),
console=console,
transient=False,
) as progress:
task = progress.add_task(
"[yellow]Processing Turtle chunks...[/yellow]",
total=None,
triples=0,
rate=0,
)
start_time = time.time()
# Estimate total chunks based on file size
estimated_chunks = max(10, int(file_size_mb * 100)) # Rough estimate
# Stream and process chunks
for _chunk_idx, chunk in enumerate(
stream_turtle_file(input_path, chunk_size)
):
# Convert chunk to Arrow table
table = pa.Table.from_pylist(chunk, schema=schema)
# Write chunk to Parquet file
chunk_file = temp_chunks_dir / f"chunk-{chunk_count:05d}.parquet"
pq.write_table(table, chunk_file)
parquet_files.append(chunk_file)
# Update progress
total_triples += len(chunk)
chunk_count += 1
elapsed = time.time() - start_time
rate = total_triples / elapsed if elapsed > 0 else 0
# Emit progress marker for wrapper scripts
progress_pct = 10 + min(70, (chunk_count / estimated_chunks) * 70)
if chunk_count % max(1, estimated_chunks // 20) == 0:
print(f"\nPROGRESS: {progress_pct:.0f}", flush=True)
progress.update(
task,
advance=1,
completed=chunk_count,
triples=total_triples,
rate=rate,
description="[yellow]Processing Turtle file...[/yellow]",
)
console.print(
f"\n[green]✓ Processed {total_triples:,} triples in "
f"{chunk_count} chunks[/green]"
)
print("\nPROGRESS: 80", flush=True)
# Merge Parquet files into final dataset structure
console.print("\n[yellow]Merging chunks into final dataset...[/yellow]")
print("\nPROGRESS: 85", flush=True)
# Read all Parquet files as a single dataset
# Note: from_parquet can return different types, but with a glob pattern
# it returns a Dataset that can be used in DatasetDict
dataset = cast(Dataset, Dataset.from_parquet(str(temp_chunks_dir / "*.parquet")))
print("\nPROGRESS: 90", flush=True)
# Create DatasetDict with single 'data' split
dataset_dict = DatasetDict({"data": dataset})
# Add metadata
if metadata:
for split in dataset_dict:
dataset_dict[split].info.description = metadata.get("description", "")
dataset_dict[split].info.citation = metadata.get("citation", "")
dataset_dict[split].info.homepage = metadata.get("homepage", "")
dataset_dict[split].info.license = metadata.get("license", "")
# Save final dataset
console.print("[yellow]Saving final dataset...[/yellow]")
print("\nPROGRESS: 95", flush=True)
dataset_dict.save_to_disk(str(output_path))
# Clean up temporary chunk files
console.print("[dim]Cleaning up temporary files...[/dim]")
shutil.rmtree(temp_chunks_dir)
print("\nPROGRESS: 100", flush=True)
# Save dataset info
elapsed_time = time.time() - start_time
info = {
"format": "parquet",
"total_triples": total_triples,
"data_size": len(dataset_dict["data"]),
"source_format": "turtle",
"chunk_size": chunk_size,
"processing_time_seconds": round(elapsed_time, 2),
"triples_per_second": round(total_triples / elapsed_time, 2)
if elapsed_time > 0
else 0,
"conversion_date": time.strftime("%Y-%m-%d %H:%M:%S"),
}
with open(output_path / "dataset_info.json", "w") as f:
json.dump(info, f, indent=2)
console.print(
"\n[bold green]✓ Successfully converted to HuggingFace dataset[/bold green]"
)
console.print(f"[green] Data split: {len(dataset_dict['data']):,} triples[/green]")
console.print(f"[green] Processing time: {elapsed_time:.1f} seconds[/green]")
console.print(f"[green] Output: {output_path}[/green]")
def main():
"""Main entry point for CLI usage."""
parser = argparse.ArgumentParser(
description=(
"Convert RDF Turtle files to HuggingFace dataset format "
"with streaming"
)
)
parser.add_argument("input", type=Path, help="Input Turtle file path")
parser.add_argument("output", type=Path, help="Output dataset directory")
parser.add_argument(
"--chunk-size",
type=int,
default=10000,
help="Number of statements to process at once (default: 10000)",
)
parser.add_argument("--description", type=str, help="Dataset description")
parser.add_argument("--homepage", type=str, help="Dataset homepage")
parser.add_argument("--license", type=str, help="Dataset license")
args = parser.parse_args()
# Validate input
if not args.input.exists():
print(f"Error: Input file '{args.input}' not found")
return 1
# Prepare metadata
metadata = {}
if args.description:
metadata["description"] = args.description
if args.homepage:
metadata["homepage"] = args.homepage
if args.license:
metadata["license"] = args.license
# Run conversion
try:
convert_turtle_to_hf_streaming(
args.input,
args.output,
chunk_size=args.chunk_size,
metadata=metadata,
)
return 0
except Exception as e:
print(f"Error: {e}")
return 1
if __name__ == "__main__":
exit(main())
scripts/convert_rdf_to_hf_dataset_unified.py
+479 -221
View File
File diff suppressed because it is too large Load Diff
scripts/dataset_downloader.py
+8 -3
View File
@@ -133,9 +133,14 @@ class DatasetDownloader:
for split_name, ds in dataset.items():
self.console.print(f" • {split_name}: {len(ds):,} rows") # type: ignore[arg-type]
else: # Single Dataset
self.console.print(
f"\n[green]✓ Downloaded dataset with {len(dataset):,} rows[/green]"
) # type: ignore[arg-type]
if isinstance(dataset, Dataset):
self.console.print(
f"\n[green]✓ Downloaded dataset with {len(dataset):,} rows[/green]"
) # type: ignore[arg-type]
else:
self.console.print(
"\n[green]✓ Downloaded dataset"
)
self.console.print(f"\n[cyan]Dataset cached at: {self.output_dir}[/cyan]")
return dataset # type: ignore[return-value]
scripts/rdf_to_hf_incremental.py
+10 -11
View File
@@ -39,9 +39,8 @@ import random
import sys
import time
import zipfile
from collections.abc import Iterator
from pathlib import Path
from typing import Iterator
from urllib.parse import urlparse
import pyarrow as pa
import pyarrow.parquet as pq
@@ -55,6 +54,8 @@ PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
from convert_rdf_to_hf_dataset_streaming_parallel import stream_rdf_chunks
from dataset_registry import DATASET_REGISTRY, DatasetInfo, get_dataset_config
from src.streaming.extract_stream import (
ExtractConfig,
_infer_name_from_url,
@@ -64,8 +65,6 @@ from src.streaming.extract_stream import (
spool_zip_to_tempfile,
)
from src.streaming.http_stream import HTTPStreamConfig, stream_http_bytes
from dataset_registry import DATASET_REGISTRY, DatasetInfo, get_dataset_config
from convert_rdf_to_hf_dataset_streaming_parallel import stream_rdf_chunks
# Setup logger
logger = logging.getLogger(__name__)
@@ -393,7 +392,7 @@ def create_rdf_chunk_iterator(
if input_type == "file":
# Local file mode: use optimized file-based streaming with multiprocessing
console.print(f"[cyan]Mode: Local file streaming[/cyan]")
console.print("[cyan]Mode: Local file streaming[/cyan]")
console.print(f"[dim]Using parallel file-based parser ({num_workers} workers)[/dim]")
file_path = Path(input_value)
chunk_iter = stream_rdf_chunks(file_path, rdf_format, DEFAULT_BATCH_SIZE, num_workers)
@@ -401,8 +400,8 @@ def create_rdf_chunk_iterator(
elif input_type == "registry":
# Streaming download mode: use line-based streaming
console.print(f"[cyan]Mode: Streaming download from registry[/cyan]")
console.print(f"[dim]Using line-based streaming parser (no multiprocessing)[/dim]")
console.print("[cyan]Mode: Streaming download from registry[/cyan]")
console.print("[dim]Using line-based streaming parser (no multiprocessing)[/dim]")
ds = DATASET_REGISTRY[input_value]
url = ds.url
@@ -665,7 +664,7 @@ def list_datasets() -> None:
by_category[info.category] = by_category.get(info.category, 0) + 1
console.print(f"\n[bold]Total:[/bold] {total} datasets ({available_count} available)")
console.print(f"[bold]By category:[/bold] ", end="")
console.print("[bold]By category:[/bold] ", end="")
console.print(", ".join(f"{cat}: {count}" for cat, count in sorted(by_category.items())))
console.print()
@@ -1021,7 +1020,7 @@ Features:
if input_type == "registry":
ds = DATASET_REGISTRY[input_value]
console.print(f"\n[bold]Dataset Info:[/bold]")
console.print("\n[bold]Dataset Info:[/bold]")
console.print(f" Name: {ds.name}")
console.print(f" Category: {ds.category}")
console.print(f" Size: {ds.size_gb} GB")
@@ -1036,12 +1035,12 @@ Features:
else:
console.print(f" Estimated triples: {ds.triples}")
console.print(f"\n[bold]Would perform:[/bold]")
console.print("\n[bold]Would perform:[/bold]")
console.print(f" 1. {'Stream from ' + DATASET_REGISTRY[input_value].url if input_type == 'registry' else 'Read from ' + input_value}")
console.print(f" 2. Parse RDF triples in {rdf_format} format")
console.print(f" 3. Convert to Parquet shards ({args.rows_per_shard:,} rows each)")
console.print(f" 4. Upload shards to {args.repo_id}")
console.print(f" 5. Create and upload README.md")
console.print(" 5. Create and upload README.md")
console.print("\n[bold green]✓ Dry run complete. Use without --dry-run to execute.[/bold green]")
return 0
scripts/upload_all_datasets.py
+21 -5
View File
@@ -55,6 +55,7 @@ DIRECTORY STRUCTURE:
from __future__ import annotations
import argparse
import os
import shutil
import subprocess
import sys
@@ -1017,14 +1018,16 @@ def convert(args: argparse.Namespace, rdf_file: Path, dataset_id: str) -> None:
/ "convert_rdf_to_hf_dataset_unified.py"
)
# Unified script will auto-select best strategy based on file size and format
# Unified script will auto-select best strategy based on file size
# and format
console.print(
"[dim]Using unified converter (auto-selects best strategy)[/dim]"
)
if args.dry_run:
console.print(
f"[dim]Would run: python scripts/convert_rdf_to_hf_dataset_unified.py "
"[dim]Would run: python "
"scripts/convert_rdf_to_hf_dataset_unified.py "
"{rdf_file} {hf_dataset_dir} --strategy auto[/dim]"
)
else:
@@ -1203,6 +1206,11 @@ def main() -> int:
"""
args, parser = parse_args()
# Fix for "AF_UNIX path too long" error in multiprocessing
brent.edwards commented 2025-12-26 22:32:00 +00:00
Review
Copy Link
Copy Source

Superb choice. Thanks.

Superb choice. Thanks.
# This forces the temporary directory to be /tmp (short path) instead of
# a potentially deep workspace path
os.environ["TMPDIR"] = "/tmp"
start_time = time.monotonic()
# List datasets to process
@@ -1255,7 +1263,9 @@ def main() -> int:
return 1
one_dataset_download_time = time.monotonic()
one_dataset_download_duration = one_dataset_download_time - one_dataset_start_time
one_dataset_download_duration = (
one_dataset_download_time - one_dataset_start_time
)
print(f"`download` took {one_dataset_download_duration:.2f} seconds:")
# Decompress
@@ -1271,10 +1281,14 @@ def main() -> int:
return 1
one_dataset_decompress_time = time.monotonic()
one_dataset_decompress_duration = one_dataset_decompress_time - one_dataset_download_time
one_dataset_decompress_duration = (
one_dataset_decompress_time - one_dataset_download_time
)
print(f"`decompress` took {one_dataset_decompress_duration:.2f} seconds:")
# Convert
assert rdf_file is not None
if not args.dry_run:
rdf_file = get_most_recent_file(rdf_file.parent)
if rdf_file is None:
@@ -1298,7 +1312,9 @@ def main() -> int:
return 1
one_dataset_convert_time = time.monotonic()
one_dataset_convert_duration = one_dataset_convert_time - one_dataset_decompress_time
one_dataset_convert_duration = (
one_dataset_convert_time - one_dataset_decompress_time
)
print(f"`convert` took {one_dataset_convert_duration:.2f} seconds:")
# Upload
shell.nix
+18
View File
@@ -0,0 +1,18 @@
let
pkgs = import <nixpkgs> {};
lib-path = with pkgs;
lib.makeLibraryPath [
stdenv.cc.cc
];
in
pkgs.mkShell {
packages = with pkgs; [
python312
(poetry.override {python3 = python312;})
uv
];
shellHook = ''
export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:${lib-path}
'';
}
src/streaming/extract_stream.py
+1 -1
View File
@@ -6,9 +6,9 @@ import io
import lzma
import tarfile
import tempfile
from collections.abc import Iterable, Iterator
from dataclasses import dataclass
from pathlib import Path
from typing import Iterable, Iterator
from urllib.parse import urlparse
src/streaming/http_stream.py
+1 -1
View File
@@ -1,8 +1,8 @@
from __future__ import annotations
import time
from collections.abc import Iterator
from dataclasses import dataclass
from typing import Iterator
import httpx