diff --git a/scripts/convert_rdf_to_hf_dataset_unified.py b/scripts/convert_rdf_to_hf_dataset_unified.py index e929bb4..aed5e89 100644 --- a/scripts/convert_rdf_to_hf_dataset_unified.py +++ b/scripts/convert_rdf_to_hf_dataset_unified.py @@ -25,6 +25,10 @@ USAGE: python convert_rdf_to_hf_dataset_unified.py input.ttl output/ \ --format turtle --description "My Dataset" --license "CC BY 4.0" + # Upload directly to HuggingFace Hub + python convert_rdf_to_hf_dataset_unified.py input.ttl \ + --format turtle --push-to-hub --hub-repo-id username/dataset-name + REQUIREMENTS: pip install rdflib datasets rich pyarrow """ @@ -67,6 +71,104 @@ from rich.progress import ( logger = logging.getLogger(__name__) +# ============================================================================= +# MODULE-LEVEL FUNCTIONS FOR MULTIPROCESSING +# ============================================================================= + +def process_geonames_lines(lines): + """Process a batch of lines from GeoNames file. + + Module-level function for multiprocessing. + + Args: + lines: List of lines containing one or more XML documents + + Returns: + List of triple dictionaries + """ + from rdflib import Graph, Literal, URIRef + + triples = [] + current_xml = [] + + for line in lines: + # 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: + triples.append(extract_triple(s, p, o)) + + except Exception as e: + # Skip malformed documents + logger.error(f"Error parsing XML document: {e}. Skipping document.") + pass + + # 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: + triples.append(extract_triple(s, p, o)) + except Exception as e: + logger.error(f"Error parsing final XML document: {e}. Skipping document.") + pass + + return triples + + +def process_ntriples_lines(lines): + """Process a batch of N-Triples lines. + + Args: + lines: List of N-Triple lines + + Returns: + List of triple dictionaries + """ + from rdflib import Graph + + triples = [] + + for line in lines: + 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: + triples.append(extract_triple(s, p, o)) + + except Exception as e: + # Skip malformed lines + logger.error(f"Error parsing N-Triple line: {line}. Skipping line. Error: {e}") + continue + + return triples + # ============================================================================= # DATA CLASSES - Configuration and Results # ============================================================================= @@ -75,7 +177,7 @@ logger = logging.getLogger(__name__) class ConversionConfig: """Configuration for RDF to HuggingFace conversion.""" input_path: Path - output_path: Path + output_path: Path | None rdf_format: str = "turtle" chunk_size: int = 10000 num_workers: int | None = None @@ -84,6 +186,8 @@ class ConversionConfig: test_size: float = 0.05 clean_cache: bool = False verbose: bool = False + push_to_hub: bool = False + hub_repo_id: str | None = None @dataclass class ConversionResult: @@ -196,6 +300,80 @@ class FileHandler: logger.error(f"Error getting file size for {file_path}: {e}") raise +def batch_file_lines(file_path: Path, batch_size: int = 1000, format: str = "geonames"): + """Generator that yields batches of lines from file. + + For GeoNames format, ensures complete XML documents are kept together. + + Args: + file_path: Path to file + batch_size: Number of lines per batch + format: File format (geonames or ntriples) + + Yields: + Batches of lines + """ + # Handle compressed files + if file_path.suffix == ".gz": + try: + file_obj = gzip.open(file_path, "rt", encoding="utf-8") + # Test read to check if file is valid + test_line = file_obj.readline() + if not test_line and file_path.stat().st_size > 0: + raise EOFError("Compressed file appears to be empty or corrupted") + file_obj.seek(0) + except (gzip.BadGzipFile, EOFError, OSError) as e: + console = Console() + console.print(f"[red]Error: Compressed file is corrupted or incomplete: {e}[/red]") + console.print(f"[yellow]File: {file_path}[/yellow]") + console.print("[yellow]Please re-download the dataset or use an uncompressed version[/yellow]") + raise + else: + file_obj = open(file_path, encoding="utf-8") + + try: + current_batch = [] + current_doc = [] + + if format == "geonames": + # For GeoNames, keep documents together + for line in file_obj: + # Check if this is a document boundary + if line.startswith("http://") or line.startswith("https://"): + if current_doc: + # Add completed document to batch + current_batch.extend(current_doc) + current_doc = [line] + + # Yield batch if large enough + if len(current_batch) >= batch_size: + yield current_batch + current_batch = [] + else: + current_doc = [line] + else: + current_doc.append(line) + + # Yield final batch + if current_doc: + current_batch.extend(current_doc) + if current_batch: + yield current_batch + + else: # ntriples or other line-based format + for line in file_obj: + current_batch.append(line) + if len(current_batch) >= batch_size: + yield current_batch + current_batch = [] + + # Yield final batch + if current_batch: + yield current_batch + + finally: + file_obj.close() + class MetadataHandler: """Handle metadata for the HuggingFace dataset.""" @staticmethod @@ -290,9 +468,18 @@ class ConversionStrategy(ABC): def _save_dataset(self, dataset_dict: DatasetDict, config: ConversionConfig) -> None: try: - config.output_path.mkdir(parents=True, exist_ok=True) - dataset_dict.save_to_disk(str(config.output_path)) - self.progress.print(f"Dataset saved to {config.output_path}", "green") + if config.push_to_hub: + if not config.hub_repo_id: + raise ValueError("hub_repo_id is required when push_to_hub is True") + self.progress.print(f"Uploading to HuggingFace Hub: {config.hub_repo_id}...", "yellow") + dataset_dict.push_to_hub(config.hub_repo_id, private=False) + self.progress.print(f"✓ Successfully uploaded to {config.hub_repo_id}", "green") + else: + if not config.output_path: + raise ValueError("output_path is required when push_to_hub is False") + config.output_path.mkdir(parents=True, exist_ok=True) + dataset_dict.save_to_disk(str(config.output_path)) + self.progress.print(f"Dataset saved to {config.output_path}", "green") except Exception as e: logger.error(f"Error saving dataset: {e}") raise @@ -383,6 +570,124 @@ class ConversionStrategy(ABC): if current_chunk: yield current_chunk + def _stream_geonames_parallel(self, config: ConversionConfig, num_workers: int) -> Iterator[list[RDFTriple]]: + """Stream GeoNames RDF file with parallel processing.""" + self.progress.print(f"[yellow]Using parallel GeoNames parser with {num_workers} workers[/yellow]") + + # Create batches of lines + batches = batch_file_lines(config.input_path, batch_size=config.chunk_size, format="geonames") + + # Process batches in parallel + try: + with mp.Pool(processes=num_workers) as pool: + for triples in pool.imap_unordered(process_geonames_lines, batches, chunksize=1): + if triples: + yield triples + except Exception as e: + logger.error(f"Error processing GeoNames file: {e}") + raise + + def _stream_ntriples_parallel(self, config: ConversionConfig, num_workers: int) -> Iterator[list[RDFTriple]]: + """Stream N-Triples file with parallel processing.""" + self.progress.print(f"[yellow]Using parallel N-Triples parser with {num_workers} workers[/yellow]") + + # Create batches of lines + batches = batch_file_lines(config.input_path, batch_size=config.chunk_size, format="ntriples") + + # Process batches in parallel + try: + with mp.Pool(processes=num_workers) as pool: + for triples in pool.imap_unordered(process_ntriples_lines, batches, chunksize=1): + if triples: + yield triples + except Exception as e: + logger.error(f"Error processing N-Triples file: {e}") + raise + + def _stream_turtle_chunks(self, config: ConversionConfig) -> Iterator[list[RDFTriple]]: + """Stream Turtle file in chunks using incremental parsing.""" + self.progress.print("[yellow]Streaming Turtle file (line-by-line parser)[/yellow]") + + # Open file with compression support + file_obj = self.file_handler.open_file(config.input_path) + + try: + current_chunk: list[str] = [] + triple_count = 0 + line_count = 0 + + # Collect prefixes first + prefix_lines: list[str] = [] + in_prefixes = True + + for line_bytes in file_obj: + line_count += 1 + # Ensure we have a string + line = line_bytes if isinstance(line_bytes, str) else line_bytes.decode('utf-8', errors='ignore') + stripped = line.strip() + + # Skip empty lines and comments + if not stripped or stripped.startswith("#"): + continue + + # Collect prefix declarations + if in_prefixes and (stripped.startswith("@prefix") or stripped.startswith("@base")): + prefix_lines.append(line) + continue + elif in_prefixes: + # End of prefixes, now we're in the data + in_prefixes = False + + # Parse in batches of lines ending with '.' + if stripped.endswith("."): + triple_count += 1 + + if triple_count >= config.chunk_size: + # Try to parse this chunk + chunk_text = "".join(prefix_lines) + "\n" + "".join(current_chunk) + "\n" + line + + try: + graph = Graph() + graph.parse(data=chunk_text, format="turtle") + + # Extract triples + triples = [extract_triple(s, p, o) for s, p, o in graph] + if triples: + yield triples + + except Exception as e: + logger.error(f"Error parsing Turtle chunk: {e}") + + # Reset for next chunk + current_chunk = [] + triple_count = 0 + else: + current_chunk.append(line) + else: + # Part of a multi-line statement + current_chunk.append(line) + + # Log progress + if line_count % 100000 == 0: + self.progress.print(f"[dim]Processed {line_count:,} lines...[/dim]") + + # Process remaining lines + if current_chunk: + chunk_text = "".join(prefix_lines) + "\n" + "".join(current_chunk) + try: + graph = Graph() + graph.parse(data=chunk_text, format="turtle") + + triples = [extract_triple(s, p, o) for s, p, o in graph] + if triples: + yield triples + + except Exception as e: + logger.error(f"Error parsing final Turtle chunk: {e}") + + finally: + file_obj.close() + # ============================================================================= # CONCRETE STRATEGIES # ============================================================================= @@ -402,6 +707,10 @@ class StandardStrategy(ConversionStrategy): file_size_mb = self.file_handler.get_file_size_mb(config.input_path) self.progress.print(f"Converting RDF file: {config.input_path}", "cyan") self.progress.print(f"File size: {file_size_mb:.2f} MB", "dim") + if config.push_to_hub: + self.progress.print(f"Destination: {config.hub_repo_id} (HuggingFace Hub)", "dim") + else: + self.progress.print(f"Output: {config.output_path}", "dim") self.progress.emit_progress(5) # Choose parsing method based on rdf_format @@ -426,7 +735,8 @@ class StandardStrategy(ConversionStrategy): self._save_dataset(dataset_dict, config) self.progress.emit_progress(100) elapsed = time.time() - start_time - self._save_dataset_info_json(config, len(triples), dataset_dict, elapsed) + if config.output_path and not config.push_to_hub: + self._save_dataset_info_json(config, len(triples), dataset_dict, elapsed) self.progress.print(f"✓ Converted {len(triples):,} triples in {elapsed:.1f}s", "bold green") return ConversionResult(success=True, total_triples=len(triples), processing_time_seconds=elapsed, @@ -455,10 +765,20 @@ class StreamingStrategy(ConversionStrategy): self.progress.print(f"Converting RDF file: {config.input_path}", "cyan") self.progress.print(f"File size: {file_size_mb:.2f} MB", "dim") self.progress.print(f"Chunk size: {config.chunk_size:,} triples", "dim") + if config.push_to_hub: + self.progress.print(f"Destination: {config.hub_repo_id} (HuggingFace Hub)", "dim") + else: + self.progress.print(f"Output: {config.output_path}", "dim") self.progress.emit_progress(5) - config.output_path.mkdir(parents=True, exist_ok=True) - temp_chunks_dir = config.output_path / "temp_chunks" - temp_chunks_dir.mkdir(exist_ok=True) + + # Create temp directory for chunks + if config.push_to_hub: + temp_chunks_dir = Path(tempfile.mkdtemp(prefix="rdf_chunks_")) + else: + config.output_path.mkdir(parents=True, exist_ok=True) + temp_chunks_dir = config.output_path / "temp_chunks" + temp_chunks_dir.mkdir(exist_ok=True) + total_triples = 0 chunk_count = 0 with self.progress.progress_bar("Processing RDF chunks...", total=None) as (progress, task): @@ -474,9 +794,14 @@ class StreamingStrategy(ConversionStrategy): self.progress.emit_progress(80) dataset_dict = self._merge_parquet_chunks(temp_chunks_dir, config, create_splits=True) shutil.rmtree(temp_chunks_dir) + + if config.push_to_hub: + self._save_dataset(dataset_dict, config) + self.progress.emit_progress(100) elapsed = time.time() - start_time - self._save_dataset_info_json(config, total_triples, dataset_dict, elapsed) + if config.output_path and not config.push_to_hub: + self._save_dataset_info_json(config, total_triples, dataset_dict, elapsed) self.progress.print("✓ Converted to HuggingFace dataset", "green") return ConversionResult(success=True, total_triples=total_triples, processing_time_seconds=elapsed, @@ -505,10 +830,20 @@ class StreamingTurtleStrategy(ConversionStrategy): self.progress.print(f"Converting Turtle: {config.input_path.name}", "cyan") self.progress.print(f"File size: {file_size_mb:.2f} MB", "dim") self.progress.print(f"Chunk size: {config.chunk_size:,} statements", "dim") + if config.push_to_hub: + self.progress.print(f"Destination: {config.hub_repo_id} (HuggingFace Hub)", "dim") + else: + self.progress.print(f"Output: {config.output_path}", "dim") self.progress.emit_progress(5) - config.output_path.mkdir(parents=True, exist_ok=True) - temp_chunks_dir = config.output_path / "temp_chunks" - temp_chunks_dir.mkdir(exist_ok=True) + + # Create temp directory for chunks + if config.push_to_hub: + temp_chunks_dir = Path(tempfile.mkdtemp(prefix="rdf_chunks_")) + else: + config.output_path.mkdir(parents=True, exist_ok=True) + temp_chunks_dir = config.output_path / "temp_chunks" + temp_chunks_dir.mkdir(exist_ok=True) + total_triples = 0 chunk_count = 0 # Use a specialized streaming method for Turtle with prefix handling. @@ -537,9 +872,14 @@ class StreamingTurtleStrategy(ConversionStrategy): self.progress.emit_progress(80) dataset_dict = self._merge_parquet_chunks(temp_chunks_dir, config) shutil.rmtree(temp_chunks_dir) + + if config.push_to_hub: + self._save_dataset(dataset_dict, config) + self.progress.emit_progress(100) elapsed = time.time() - start_time - self._save_dataset_info_json(config, total_triples, dataset_dict, elapsed) + if config.output_path and not config.push_to_hub: + self._save_dataset_info_json(config, total_triples, dataset_dict, elapsed) self.progress.print("✓ Converted to HuggingFace dataset", "green") return ConversionResult(success=True, total_triples=total_triples, processing_time_seconds=elapsed, @@ -568,9 +908,19 @@ class SimpleStreamingStrategy(ConversionStrategy): self.progress.print(f"Converting: {config.input_path.name}", "cyan") self.progress.print(f"File size: {file_size_mb:.2f} MB", "dim") self.progress.print(f"Chunk size: {config.chunk_size:,} triples", "dim") - config.output_path.mkdir(parents=True, exist_ok=True) - temp_chunks_dir = config.output_path / "temp_chunks" - temp_chunks_dir.mkdir(exist_ok=True) + if config.push_to_hub: + self.progress.print(f"Destination: {config.hub_repo_id} (HuggingFace Hub)", "dim") + else: + self.progress.print(f"Output: {config.output_path}", "dim") + + # Create temp directory for chunks + if config.push_to_hub: + temp_chunks_dir = Path(tempfile.mkdtemp(prefix="rdf_chunks_")) + else: + config.output_path.mkdir(parents=True, exist_ok=True) + temp_chunks_dir = config.output_path / "temp_chunks" + temp_chunks_dir.mkdir(exist_ok=True) + total_triples = 0 chunk_count = 0 with self.progress.progress_bar("Processing simple streaming chunks...", total=None) as (progress, task): @@ -585,8 +935,13 @@ class SimpleStreamingStrategy(ConversionStrategy): self.progress.print(f"✓ Processed {total_triples:,} triples in {chunk_count} chunks", "green") dataset_dict = self._merge_parquet_chunks(temp_chunks_dir, config) shutil.rmtree(temp_chunks_dir) + + if config.push_to_hub: + self._save_dataset(dataset_dict, config) + elapsed = time.time() - start_time - self._save_dataset_info_json(config, total_triples, dataset_dict, elapsed) + if config.output_path and not config.push_to_hub: + self._save_dataset_info_json(config, total_triples, dataset_dict, elapsed) self.progress.print("✓ Converted to HuggingFace dataset", "green") return ConversionResult(success=True, total_triples=total_triples, processing_time_seconds=elapsed, @@ -606,50 +961,129 @@ class ParallelStreamingStrategy(ConversionStrategy): @property def description(self) -> str: - return "Parallel streaming conversion using multiprocessing." + return "Parallel streaming conversion using multiprocessing with Dataset.from_generator()." def convert(self, config: ConversionConfig) -> ConversionResult: start_time = time.time() - num_workers = config.num_workers or mp.cpu_count() + num_workers = config.num_workers or max(1, mp.cpu_count() - 1) try: file_size_mb = self.file_handler.get_file_size_mb(config.input_path) self.progress.print(f"Converting: {config.input_path.name}", "cyan") self.progress.print(f"File size: {file_size_mb:.2f} MB", "dim") - self.progress.print(f"Chunk size: {config.chunk_size:,} lines", "dim") + self.progress.print(f"Chunk size: {config.chunk_size:,} lines per batch", "dim") self.progress.print(f"Workers: {num_workers} CPU cores", "dim") + if config.push_to_hub: + self.progress.print(f"Destination: {config.hub_repo_id} (HuggingFace Hub)", "dim") + else: + self.progress.print(f"Output: {config.output_path}", "dim") self.progress.emit_progress(5) - config.output_path.mkdir(parents=True, exist_ok=True) - temp_chunks_dir = config.output_path / "temp_chunks" - temp_chunks_dir.mkdir(exist_ok=True) - total_triples = 0 - chunk_count = 0 + + # Check if this is GeoNames format + is_geonames = "geonames" in str(config.input_path).lower() + + self.progress.print("\n[yellow]Creating dataset from streamed chunks...[/yellow]") + self.progress.emit_progress(10) + + # Estimate total chunks based on file size (rough estimate) estimated_chunks = max(10, int(file_size_mb * 1024 * 1024 / (config.chunk_size * 100))) - with self.progress.progress_bar("Processing parallel chunks...", total=None) as (progress, task): - # For parallel processing, choose method based on format - if config.rdf_format in ("nt", "ntriples"): - batches = list(self._stream_ntriples(config)) + + # Create generator function that yields individual triples + def dataset_generator(): + triple_count = 0 + chunk_count = 0 + last_print_count = 0 + + # Select appropriate streaming method based on format + if is_geonames and config.rdf_format in ("xml", "application/rdf+xml"): + # Check if it's the special GeoNames format (URLs followed by XML) + with open(config.input_path, encoding="utf-8", errors="ignore") as f: + first_line = f.readline().strip() + if first_line.startswith("http://") or first_line.startswith("https://"): + chunk_iter = self._stream_geonames_parallel(config, num_workers) + else: + chunk_iter = self._stream_generic_rdf(config) + elif config.rdf_format in ("nt", "ntriples"): + chunk_iter = self._stream_ntriples_parallel(config, num_workers) + elif config.rdf_format in ("turtle", "ttl"): + chunk_iter = self._stream_turtle_chunks(config) else: - batches = list(self._stream_generic_rdf(config)) - # Use multiprocessing Pool for parallel processing the batches - with mp.Pool(processes=num_workers) as pool: - results = pool.map(lambda b: b, batches) - for batch in results: - schema = self.schema_manager.get_arrow_schema() - table = pa.Table.from_pylist(batch, schema=schema) - chunk_file = temp_chunks_dir / f"chunk-{chunk_count:05d}.parquet" - pq.write_table(table, chunk_file) - total_triples += len(batch) - chunk_count += 1 - progress.update(task, advance=1) - self.progress.print(f"✓ Processed {total_triples:,} triples in {chunk_count} chunks", "green") - self.progress.emit_progress(80) - dataset_dict = self._merge_parquet_chunks(temp_chunks_dir, config) - shutil.rmtree(temp_chunks_dir) - self.progress.emit_progress(100) + # For other formats, use generic streaming + chunk_iter = self._stream_generic_rdf(config) + + # Yield individual triples from chunks + for chunk in chunk_iter: + chunk_count += 1 + for triple in chunk: + triple_count += 1 + yield triple + + if triple_count - last_print_count >= 100000: + elapsed = time.time() - start_time + rate = triple_count / elapsed if elapsed > 0 else 0 + progress_pct = 10 + min(60, int((chunk_count / estimated_chunks) * 60)) + self.progress.print( + f" Processing: {chunk_count:,} chunks • " + f"{triple_count:,} triples • {rate:.0f} triples/sec", + ) + self.progress.emit_progress(progress_pct) + last_print_count = triple_count + + # Create dataset using from_generator for true streaming + features = self.schema_manager.get_hf_features() + dataset = Dataset.from_generator(dataset_generator, features=features) + + self.progress.emit_progress(75) + self.progress.print("[green]✓ Dataset generation complete[/green]") + + # Get statistics after generation + total_triples = len(dataset) + + # Wrap in DatasetDict + dataset_dict = DatasetDict({"data": dataset}) + + # Add metadata + if config.metadata: + dataset_dict = self.metadata_handler.add_metadata(dataset_dict, config.metadata) + elapsed = time.time() - start_time - self._save_dataset_info_json(config, total_triples, dataset_dict, elapsed, - extra_info={"num_workers": num_workers}) - self.progress.print("✓ Converted to HuggingFace dataset", "green") + + if config.push_to_hub: + self.progress.print(f"[yellow]Uploading to HuggingFace Hub: {config.hub_repo_id}...[/yellow]") + self.progress.emit_progress(95) + dataset_dict.push_to_hub(config.hub_repo_id, private=False) + self.progress.print(f"[bold green]✓ Successfully uploaded to {config.hub_repo_id}[/bold green]") + self.progress.emit_progress(100) + else: + self.progress.print("[yellow]Saving final dataset...[/yellow]") + self.progress.emit_progress(95) + config.output_path.mkdir(parents=True, exist_ok=True) + dataset_dict.save_to_disk(str(config.output_path), num_proc=num_workers) + self.progress.emit_progress(100) + + # Save dataset info + info = { + "format": "parquet", + "total_triples": total_triples, + "data_size": len(dataset_dict["data"]), + "source_format": config.rdf_format, + "chunk_size": config.chunk_size, + "num_workers": num_workers, + "processing_time_seconds": round(elapsed, 2), + "triples_per_second": round(total_triples / elapsed, 2) if elapsed > 0 else 0, + "conversion_date": time.strftime("%Y-%m-%d %H:%M:%S"), + } + with open(config.output_path / "dataset_info.json", "w") as f: + json.dump(info, f, indent=2) + + self.progress.print("\n[bold green]✓ Successfully converted to HuggingFace dataset[/bold green]") + self.progress.print(f"[green] Data split: {len(dataset_dict['data']):,} triples[/green]") + self.progress.print(f"[green] Processing time: {elapsed:.1f} seconds[/green]") + self.progress.print(f"[green] Speed: {total_triples / elapsed:.0f} triples/second[/green]") + if config.push_to_hub: + self.progress.print(f"[green] Repository: {config.hub_repo_id}[/green]") + else: + self.progress.print(f"[green] Location: {config.output_path}[/green]") + return ConversionResult(success=True, total_triples=total_triples, processing_time_seconds=elapsed, output_path=config.output_path, @@ -725,7 +1159,7 @@ Strategies: """ ) 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("output", nargs="?", type=Path, help="Output directory for HuggingFace dataset (optional if --push-to-hub)") parser.add_argument("-f", "--format", default="turtle", choices=["turtle", "nt", "ntriples", "xml", "n3", "trig", "nquads", "tsv"], help="RDF format (default: turtle)") @@ -747,10 +1181,20 @@ Strategies: parser.add_argument("--citation", type=str, help="Dataset citation") parser.add_argument("--homepage", type=str, help="Dataset homepage URL") parser.add_argument("--license", type=str, help="Dataset license") + parser.add_argument("--push-to-hub", action="store_true", + help="Upload directly to HuggingFace Hub without saving to disk") + parser.add_argument("--hub-repo-id", type=str, + help="HuggingFace Hub repository ID (required if --push-to-hub)") parser.add_argument("-v", "--verbose", action="store_true", help="Verbose output including warnings") args = parser.parse_args() + # Validate arguments + if args.push_to_hub and not args.hub_repo_id: + parser.error("--hub-repo-id is required when using --push-to-hub") + if not args.push_to_hub and not args.output: + parser.error("output directory is required when not using --push-to-hub") + log_level = logging.WARNING if args.verbose else logging.ERROR logging.basicConfig(level=log_level, format="%(levelname)s:%(name)s:%(message)s", force=True) logging.getLogger("rdflib").setLevel(log_level) @@ -789,14 +1233,19 @@ Strategies: create_train_test_split=args.create_splits, test_size=args.test_size, clean_cache=args.clean_cache, - verbose=args.verbose + verbose=args.verbose, + push_to_hub=args.push_to_hub, + hub_repo_id=args.hub_repo_id ) progress.print("\n" + "="*60, "cyan") progress.print("Unified RDF to HuggingFace Dataset Converter", "bold cyan") progress.print("="*60, "cyan") progress.print(f"Input: {config.input_path}", "") - progress.print(f"Output: {config.output_path}", "") + if config.push_to_hub: + progress.print(f"Destination: {config.hub_repo_id} (HuggingFace Hub)", "") + else: + progress.print(f"Output: {config.output_path}", "") progress.print(f"Format: {config.rdf_format}", "") progress.print(f"Strategy: {strategy.name}", "") progress.print("", "") @@ -808,7 +1257,10 @@ Strategies: progress.print("="*60, "green") progress.print(f"Total triples: {result.total_triples:,}", "green") progress.print(f"Processing time: {result.processing_time_seconds:.1f}s", "green") - progress.print(f"Output: {result.output_path}", "green") + if config.push_to_hub: + progress.print(f"Repository: {config.hub_repo_id}", "green") + else: + progress.print(f"Output: {result.output_path}", "green") return 0 else: progress.print("\n" + "="*60, "red") diff --git a/scripts/upload_all_datasets.py b/scripts/upload_all_datasets.py index 6b90892..65b4b6a 100755 --- a/scripts/upload_all_datasets.py +++ b/scripts/upload_all_datasets.py @@ -731,11 +731,11 @@ def get_rdf_format(dataset_info: DatasetInfo) -> str: # Get format mappings from config config = get_dataset_config() format_map = config.get("format_mappings", {}) - + # Default mappings for common formats default_mappings = { "rdf": "xml", - "rdf/xml": "xml", + "rdf/xml": "xml", "xml": "xml", "turtle": "turtle", "ttl": "turtle", @@ -745,18 +745,18 @@ def get_rdf_format(dataset_info: DatasetInfo) -> str: "jsonld": "json-ld", "json-ld": "json-ld" } - + format_lower = dataset_info.format.lower() - + # First try config mappings, then default mappings, then fallback mapped_format = format_map.get(format_lower) if mapped_format: return mapped_format - + mapped_format = default_mappings.get(format_lower) if mapped_format: return mapped_format - + # Fallback to turtle for unknown formats return "turtle" @@ -831,10 +831,11 @@ This dataset was prepared using the CleverErnie GISM framework: python scripts/rdf_dataset_downloader.py {dataset_info.id} -o datasets/ # Convert to HuggingFace format -python scripts/convert_rdf_to_hf_dataset.py \\ +python scripts/convert_rdf_to_hf_dataset_unified.py \\ datasets/{dataset_info.id}/[file] \\ hf_datasets/{dataset_info.id} \\ - --format {get_rdf_format(dataset_info)} + --format {get_rdf_format(dataset_info)} \\ + --strategy auto # Upload to HuggingFace Hub python scripts/upload_all_datasets.py --dataset {dataset_info.id} @@ -849,8 +850,9 @@ python scripts/upload_all_datasets.py --dataset {dataset_info.id} ### Conversion Details - Converted using: [CleverErnie GISM](https://github.com/cleverthis/cleverernie) -- Conversion script: `scripts/convert_rdf_to_hf_dataset.py` +- Conversion script: `scripts/convert_rdf_to_hf_dataset_unified.py` - Dataset format: Single 'data' split with all triples +- Strategy: Auto-selected based on dataset size and format ### Maintenance @@ -1009,85 +1011,52 @@ def convert(args: argparse.Namespace, rdf_file: Path, dataset_id: str) -> None: console.print(f"[green]Created dataset card: {readme_path}[/green]") else: # Original conversion logic for other datasets - # Determine if we should use streaming converter based on dataset size - use_streaming = False - streaming_script = ( + # Use unified converter which automatically selects best strategy + unified_script = ( Path(__file__).parent - / "convert_rdf_to_hf_dataset_streaming_parallel.py" + / "convert_rdf_to_hf_dataset_unified.py" ) - # Use streaming for medium and larger datasets, or specifically for geonames - if ( - dataset_info.category in ["medium", "large", "xlarge"] - or "geonames" in dataset_id.lower() - ): - use_streaming = streaming_script.exists() - if use_streaming: - console.print( - "[dim]Using parallel streaming converter for " - f"{dataset_info.category} dataset[/dim]" - ) + # 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: - script_name = ( - "convert_rdf_to_hf_dataset_streaming_parallel.py" - if use_streaming - else "convert_rdf_to_hf_dataset.py" - ) console.print( - f"[dim]Would run: python scripts/{script_name} " - "{rdf_file} {hf_dataset_dir}[/dim]" + f"[dim]Would run: python scripts/convert_rdf_to_hf_dataset_unified.py " + "{rdf_file} {hf_dataset_dir} --strategy auto[/dim]" ) else: - # Choose appropriate converter script - if use_streaming: - # Use streaming converter with the appropriate chunk size - # based on dataset size. Larger chunk sizes are more memory - # efficient but still reasonable - chunk_size = 50000 # Default for streaming - if dataset_info.size_gb > 10: - chunk_size = 100000 # Larger chunks for very large datasets - elif dataset_info.size_gb < 1: - chunk_size = 10000 # Smaller chunks for small datasets + # Use unified converter with auto-strategy selection + # Adjust chunk size based on dataset size + chunk_size = 50000 # Default + if dataset_info.size_gb > 10: + chunk_size = 100000 # Larger chunks for very large datasets + elif dataset_info.size_gb < 1: + chunk_size = 10000 # Smaller chunks for small datasets - result = subprocess.run( - [ - sys.executable, - str(streaming_script), - str(rdf_file), - str(hf_dataset_dir), - "--format", - get_rdf_format(dataset_info), - "--chunk-size", - str(chunk_size), - "--description", - dataset_info.description, - "--homepage", - dataset_info.url, - "--license", - dataset_info.license if dataset_info.license else "Unknown", - ], - capture_output=False, - ) - else: - # Use original converter for small datasets - result = subprocess.run( - [ - sys.executable, - str(Path(__file__).parent / "convert_rdf_to_hf_dataset.py"), - str(rdf_file), - str(hf_dataset_dir), - "--format", - get_rdf_format(dataset_info), - "--description", - dataset_info.description, - "--homepage", - dataset_info.url, - "--license", - dataset_info.license if dataset_info.license else "Unknown", - ], - capture_output=False, - ) + result = subprocess.run( + [ + sys.executable, + str(unified_script), + str(rdf_file), + str(hf_dataset_dir), + "--format", + get_rdf_format(dataset_info), + "--strategy", + "auto", # Let the unified script auto-select best strategy + "--chunk-size", + str(chunk_size), + "--description", + dataset_info.description, + "--homepage", + dataset_info.url, + "--license", + dataset_info.license if dataset_info.license else "Unknown", + ], + capture_output=False, + ) if result.returncode != 0: console.print( @@ -1290,13 +1259,16 @@ def main() -> int: print(f"`download` took {one_dataset_download_duration:.2f} seconds:") # Decompress - console.print(f"[bold cyan]Decompressing {dataset_id}...[/bold cyan]") + if args.dry_run: + console.print(f"[dim][DRY RUN] Would decompress {dataset_id}...[/dim]") + else: + console.print(f"[bold cyan]Decompressing {dataset_id}...[/bold cyan]") - assert rdf_file is not None - result = decompress(rdf_file) - if not result: - console.print("[red]Decompression failed. Aborting.[/red]") - return 1 + assert rdf_file is not None + result = decompress(rdf_file) + if not result: + console.print("[red]Decompression failed. Aborting.[/red]") + return 1 one_dataset_decompress_time = time.monotonic() one_dataset_decompress_duration = one_dataset_decompress_time - one_dataset_download_time