From 6ff2719a8e237056c84c6a36dc8d9aaefaf753c2 Mon Sep 17 00:00:00 2001 From: Aditya Chhabra Date: Mon, 5 Jan 2026 19:03:55 +0530 Subject: [PATCH 1/4] feat: add streaming of rdf xml dataset with resume functionality --- scripts/rdf_to_hf_incremental.py | 266 +++++++++++++++++++++++++++++-- src/streaming/http_stream.py | 94 +++++++++++ 2 files changed, 345 insertions(+), 15 deletions(-) diff --git a/scripts/rdf_to_hf_incremental.py b/scripts/rdf_to_hf_incremental.py index e5558cf..24b9211 100644 --- a/scripts/rdf_to_hf_incremental.py +++ b/scripts/rdf_to_hf_incremental.py @@ -37,6 +37,7 @@ import logging import multiprocessing as mp import os import random +import re import sys import time import zipfile @@ -227,6 +228,213 @@ def stream_ntriples_from_lines(lines: Iterator[str]) -> Iterator[list[dict]]: yield batch +def _wrap_description_xml( + xml_declaration: str | None, + rdf_root: str | None, + description_lines: list[str], +) -> str: + """Wrap a Description element in a minimal valid XML document. + + Args: + xml_declaration: XML declaration line () + rdf_root: Root RDF element with namespaces () + description_lines: Lines making up the Description element + + Returns: + Complete XML document string + """ + if xml_declaration is None: + xml_declaration = '' + + if rdf_root is None: + # Fallback minimal namespace declaration + rdf_root = '' + + # Ensure rdf_root is properly closed for wrapping + if not rdf_root.rstrip().endswith(">"): + rdf_root = rdf_root.rstrip() + ">" + + description_text = "".join(description_lines) + + # Construct minimal valid XML document + wrapped = f"{xml_declaration}\n{rdf_root}\n{description_text}\n" + + return wrapped + + +def _parse_description_xml(xml_content: str) -> list[dict]: + """Parse a Description XML fragment and extract all triples. + + Args: + xml_content: Complete XML document containing one Description element + + Returns: + List of triple dictionaries + """ + try: + graph = Graph() + graph.parse(data=xml_content, format="xml") + + # Extract all triples from this Description + triples = [_convert_rdf_triple_to_dict(s, p, o) for s, p, o in graph] + return triples + except Exception as e: + logger.error(f"Error parsing Description XML: {e}") + logger.debug(f"XML content (first 500 chars): {xml_content[:500]}") + return [] # Return empty list on error, don't crash + + +def stream_rdfxml_chunks_from_lines( + lines: Iterator[str], + chunk_size: int = DEFAULT_BATCH_SIZE, +) -> Iterator[list[dict]]: + """Stream RDF/XML format from a line iterator. + + Strategy: + 1. Extract root element with namespaces (first 2 lines) + 2. Track state: inside/outside Description element + 3. Accumulate lines until is found + 4. Wrap each Description in minimal XML document with namespaces + 5. Parse each Description separately + 6. Extract all triples from each Description + + Args: + lines: Iterator yielding text lines of RDF/XML data + chunk_size: Number of Description elements to accumulate before yielding + + Yields: + Lists of triple dictionaries + """ + console = Console() + console.print("[yellow]Streaming RDF/XML from line iterator[/yellow]") + + # State tracking + root_line: str | None = None # XML declaration + rdf_root_line: str | None = None # with namespaces + current_description: list[str] = [] + in_description = False + description_count = 0 + triple_batch: list[dict] = [] + line_count = 0 + + # Track XML element depth for nested elements + depth = 0 + + for line in lines: + line_count += 1 + stripped = line.strip() + + # Skip empty lines + if not stripped: + continue + + # Extract XML declaration (line 1) + if stripped.startswith(" + if stripped.rstrip().endswith("/>"): + # Self-closing, process immediately + in_description = False + depth = 0 + description_xml = _wrap_description_xml( + root_line, rdf_root_line, current_description + ) + triples = _parse_description_xml(description_xml) + triple_batch.extend(triples) + description_count += 1 + current_description = [] + + if description_count >= chunk_size: + yield triple_batch + triple_batch = [] + description_count = 0 + continue + + # If we're inside a Description, accumulate lines + if in_description: + current_description.append(line) + + # Count opening/closing tags to track depth correctly + # Opening tags: or (not ) + # Find all because regex requires [a-zA-Z_] after <) + all_tag_starts = re.findall(r'<([a-zA-Z_][\w:.-]*)', stripped) + # Closing tags: + closing_tags = len(re.findall(r'', stripped)) + # Self-closing tags: (these don't change depth) + # Find tags that end with /> before the next > + self_closing_tags = len(re.findall(r'<([a-zA-Z_][\w:.-]*)[^>]*/>', stripped)) + + # Opening tags = all tag starts - self-closing tags + # (all_tag_starts already excludes closing tags, so we only subtract self-closing) + opening_tags = len(all_tag_starts) - self_closing_tags + + # Net depth change: opening tags increase depth, closing tags decrease it + # Self-closing tags don't change depth (already excluded from opening_tags) + depth += opening_tags - closing_tags + + # Check if Description element is closed + if "" in stripped: + # The closing tag was already accounted for in closing_tags above + if depth == 0: + # Complete Description element found + in_description = False + + # Wrap in minimal XML document + description_xml = _wrap_description_xml( + root_line, rdf_root_line, current_description + ) + + # Parse and extract triples + triples = _parse_description_xml(description_xml) + triple_batch.extend(triples) + description_count += 1 + + # Reset for next Description + current_description = [] + + # Yield when chunk_size reached + if description_count >= chunk_size: + yield triple_batch + triple_batch = [] + description_count = 0 + + # Handle other elements (like owl:Ontology) - skip for now + # Could be extended to parse these too if needed + + if line_count % PROGRESS_LOG_INTERVAL == 0: + console.print( + f"[dim]Processed {line_count:,} lines, {description_count:,} descriptions...[/dim]" + ) + + # Process final Description if file ends mid-element + if current_description and in_description: + # Try to parse anyway (might be incomplete) + try: + description_xml = _wrap_description_xml( + root_line, rdf_root_line, current_description + ) + triples = _parse_description_xml(description_xml) + triple_batch.extend(triples) + except Exception as e: + logger.warning(f"Incomplete Description at end of file: {e}") + + # Yield final batch + if triple_batch: + yield triple_batch + + def stream_rdf_chunks_from_lines( lines: Iterator[str], format: str = "turtle", @@ -236,7 +444,7 @@ def stream_rdf_chunks_from_lines( Args: lines: Iterator yielding text lines of RDF data - format: RDF serialization format (supported: nt, ntriples, turtle, ttl) + format: RDF serialization format (supported: nt, ntriples, turtle, ttl, xml, rdf, rdfxml) chunk_size: Number of triples per chunk for Turtle parsing Yields: @@ -249,10 +457,12 @@ def stream_rdf_chunks_from_lines( yield from stream_ntriples_from_lines(lines) elif format in ("turtle", "ttl"): yield from stream_turtle_chunks_from_lines(lines, chunk_size=chunk_size) + elif format in ("xml", "rdf", "rdfxml"): + yield from stream_rdfxml_chunks_from_lines(lines, chunk_size=chunk_size) else: raise ValueError( f"Streaming line-iterator parser not implemented for format: {format}. " - f"Supported formats: nt, ntriples, turtle, ttl" + f"Supported formats: nt, ntriples, turtle, ttl, xml, rdf, rdfxml" ) @@ -261,9 +471,11 @@ def stream_rdf_chunks_from_lines( # ============================================================================ -def _pick_zip_member(zf: zipfile.ZipFile, - *, - prefer_exts: tuple[str, ...] = (".ttl", ".nt")) -> str: +def _pick_zip_member( + zf: zipfile.ZipFile, + *, + prefer_exts: tuple[str, ...] = (".ttl", ".nt", ".rdf", ".xml") +) -> str: """Select the best RDF member file from a ZIP archive.""" candidates = [] for info in zf.infolist(): @@ -315,14 +527,14 @@ def open_rdf_line_stream_from_url( # TAR archives if lower.endswith((".tar.gz", ".tgz", ".tar")): for member_name, member_bytes in iter_tar_member_bytes(name, byte_iter): - if member_name.lower().endswith((".ttl", ".nt")): + if member_name.lower().endswith((".ttl", ".nt", ".rdf", ".xml")): lines = iter_text_lines( member_bytes, encoding=extract_cfg.encoding, errors=extract_cfg.errors ) return lines, f"{name}:{member_name}" - raise ValueError("No .ttl/.nt member found in tar archive stream.") + raise ValueError("No .ttl/.nt/.rdf/.xml member found in tar archive stream.") # ZIP archives if lower.endswith(".zip"): @@ -1028,24 +1240,48 @@ Features: if not rdf_format: if input_type == "registry": ds = DATASET_REGISTRY[input_value] - rdf_format = ds.format - console.print("[dim]Format auto-detected from registry: " - f"{rdf_format}[/dim]") + # Map registry format to streaming parser format + registry_format = ds.format.lower() + format_mapping = { + "rdf": "rdf", + "rdf/xml": "rdf", + "xml": "rdf", + "turtle": "turtle", + "ttl": "turtle", + "nt": "ntriples", + "ntriples": "ntriples", + } + rdf_format = format_mapping.get(registry_format, registry_format) + console.print( + f"[dim]Format auto-detected from registry: " + f"{registry_format} → {rdf_format}[/dim]" + ) elif input_type == "file": - # Infer from file extension, default to turtle - ext = Path(input_value).suffix.lower() + # Infer from file extension, handling compressed files + file_path = Path(input_value) + # Get stem (filename without extension) and check for double extensions + stem = file_path.stem.lower() + ext = file_path.suffix.lower() + + # Handle compressed files: .rdf.xz, .ttl.gz, etc. + if ext in (".gz", ".bz2", ".xz", ".zip"): + # Check if stem has an RDF extension + stem_path = Path(stem) + if stem_path.suffix.lower() in (".rdf", ".xml", ".ttl", ".nt"): + ext = stem_path.suffix.lower() + format_map = { ".ttl": "turtle", ".nt": "ntriples", ".nq": "nquads", - ".rdf": "xml", - ".xml": "xml", + ".rdf": "rdf", + ".xml": "rdf", ".jsonld": "json-ld", ".n3": "n3", ".trig": "trig" } rdf_format = format_map.get(ext, "turtle") - console.print(f"[dim]Format inferred from extension: {rdf_format}[/dim]") + console.print(f"[dim]Format inferred from extension: {ext} → {rdf_format}[/dim]") console.print(f"[dim]Rows per shard: {args.rows_per_shard:,}, " f"Record batch size: {args.record_batch_size:,}[/dim]") diff --git a/src/streaming/http_stream.py b/src/streaming/http_stream.py index 81276d0..a65254d 100644 --- a/src/streaming/http_stream.py +++ b/src/streaming/http_stream.py @@ -1,8 +1,11 @@ from __future__ import annotations import time +import urllib.error +import urllib.request from collections.abc import Iterator from dataclasses import dataclass +from urllib.parse import urlparse import httpx @@ -24,6 +27,31 @@ def stream_http_bytes( headers: dict[str, str] | None = None, start_byte: int | None = None, config: HTTPStreamConfig | None = None, +) -> Iterator[bytes]: + """Stream bytes from an HTTP(S) or FTP URL with retry and optional Range start. + + - Yields chunks of bytes. + - If start_byte is provided, attempts to use Range requests (HTTP/HTTPS only). + - Retries on network/timeouts and 5xx responses with exponential backoff. + - Automatically detects protocol and routes to appropriate handler. + """ + parsed = urlparse(url) + protocol = parsed.scheme.lower() + + if protocol == "ftp": + yield from stream_ftp_bytes(url, config=config) + elif protocol in ("http", "https"): + yield from stream_http_bytes_impl(url, headers=headers, start_byte=start_byte, config=config) + else: + raise ValueError(f"Unsupported protocol: {protocol}. Supported: http, https, ftp") + + +def stream_http_bytes_impl( + url: str, + *, + headers: dict[str, str] | None = None, + start_byte: int | None = None, + config: HTTPStreamConfig | None = None, ) -> Iterator[bytes]: """Stream bytes from an HTTP(S) URL with retry and optional Range start. @@ -77,3 +105,69 @@ def stream_http_bytes( time.sleep(sleep_s) +def stream_ftp_bytes( + url: str, + *, + config: HTTPStreamConfig | None = None, +) -> Iterator[bytes]: + """Stream bytes from an FTP URL with retry logic. + + FTP Protocol Limitations: + - No Range header support (cannot resume interrupted downloads) + - No Content-Length in all cases (progress may be indeterminate) + + Args: + url: FTP URL to stream from + config: Streaming configuration (chunk_size, timeouts, retries) + + Yields: + Chunks of bytes from the FTP stream + + Raises: + ValueError: If FTP authentication fails or file not found + urllib.error.URLError: On network errors after retries exhausted + """ + cfg = config or HTTPStreamConfig() + attempt = 0 + + while True: + attempt += 1 + try: + timeout = int(cfg.connect_timeout_s + cfg.read_timeout_s) + with urllib.request.urlopen(url, timeout=timeout) as response: + while True: + chunk = response.read(cfg.chunk_size) + if not chunk: + break + yield chunk + return + + except urllib.error.URLError as e: + reason = str(e.reason) if hasattr(e, "reason") else str(e) + + # FTP-specific error codes + if "530" in reason or "Login incorrect" in reason: + raise ValueError(f"FTP authentication failed: {url}") from e + elif "550" in reason or "No such file" in reason: + raise ValueError(f"File not found on FTP server: {url}") from e + elif "timed out" in reason.lower(): + # Retry on timeout + if attempt > cfg.max_retries: + raise + sleep_s = min(cfg.backoff_base_s * (2 ** (attempt - 1)), cfg.backoff_max_s) + time.sleep(sleep_s) + else: + # Other errors: retry if we have attempts left + if attempt > cfg.max_retries: + raise + sleep_s = min(cfg.backoff_base_s * (2 ** (attempt - 1)), cfg.backoff_max_s) + time.sleep(sleep_s) + + except Exception as e: + # Unexpected errors: retry if we have attempts left + if attempt > cfg.max_retries: + raise + sleep_s = min(cfg.backoff_base_s * (2 ** (attempt - 1)), cfg.backoff_max_s) + time.sleep(sleep_s) + + -- 2.52.0 From 9ccdd38e91a75ac82f6d7529d4c24f9ad2d0048c Mon Sep 17 00:00:00 2001 From: Aditya Chhabra Date: Tue, 6 Jan 2026 19:35:48 +0530 Subject: [PATCH 2/4] fix: parse xml file with xml.etree.ElementTree; fix ruff and pyright errors --- scripts/rdf_to_hf_incremental.py | 506 ++++++++++++++++--------------- src/streaming/http_stream.py | 51 +++- 2 files changed, 302 insertions(+), 255 deletions(-) diff --git a/scripts/rdf_to_hf_incremental.py b/scripts/rdf_to_hf_incremental.py index 24b9211..f699a11 100644 --- a/scripts/rdf_to_hf_incremental.py +++ b/scripts/rdf_to_hf_incremental.py @@ -32,27 +32,38 @@ from __future__ import annotations import argparse import contextlib import hashlib +import io import json import logging import multiprocessing as mp import os import random -import re import sys import time import zipfile from collections.abc import Iterator from pathlib import Path +from xml.etree import ElementTree as ET import pyarrow as pa import pyarrow.parquet as pq -from convert_rdf_to_hf_dataset_streaming_parallel import stream_rdf_chunks -from dataset_registry import DATASET_REGISTRY, DatasetInfo, get_dataset_config from huggingface_hub import HfApi, hf_hub_download from rdflib import Graph, Literal, URIRef from rich.console import Console from rich.table import Table -from src.streaming.extract_stream import ( + +# Ensure project root is on sys.path +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 # noqa: E402 +from dataset_registry import ( # noqa: E402 + DATASET_REGISTRY, + DatasetInfo, + get_dataset_config, +) +from src.streaming.extract_stream import ( # noqa: E402 ExtractConfig, _infer_name_from_url, iter_decompressed_bytes, @@ -61,12 +72,7 @@ from src.streaming.extract_stream import ( spool_zip_to_tempfile, detect_compression_format, ) -from src.streaming.http_stream import HTTPStreamConfig, stream_http_bytes - -# Ensure project root is on sys.path -PROJECT_ROOT = Path(__file__).resolve().parents[1] -if str(PROJECT_ROOT) not in sys.path: - sys.path.insert(0, str(PROJECT_ROOT)) +from src.streaming.http_stream import HTTPStreamConfig, stream_http_bytes # noqa: E402 # Setup logger @@ -147,8 +153,8 @@ def stream_turtle_chunks_from_lines( if not stripped or stripped.startswith("#"): continue - if (in_prefixes - and (stripped.startswith("@prefix") or stripped.startswith("@base"))): + is_prefix = stripped.startswith("@prefix") or stripped.startswith("@base") + if in_prefixes and is_prefix: prefix_lines.append(line) continue elif in_prefixes: @@ -159,17 +165,16 @@ def stream_turtle_chunks_from_lines( if triple_count >= chunk_size: chunk_text = ( - "".join(prefix_lines) - + "\n" - + "".join(current_chunk) - + "\n" - + line + "".join(prefix_lines) + "\n" + + "".join(current_chunk) + "\n" + line ) try: graph = Graph() graph.parse(data=chunk_text, format="turtle") - triples = [_convert_rdf_triple_to_dict(s, p, o) - for s, p, o in graph] + triples = [ + _convert_rdf_triple_to_dict(s, p, o) + for s, p, o in graph + ] if triples: yield triples except Exception as e: @@ -228,37 +233,58 @@ def stream_ntriples_from_lines(lines: Iterator[str]) -> Iterator[list[dict]]: yield batch -def _wrap_description_xml( - xml_declaration: str | None, - rdf_root: str | None, - description_lines: list[str], -) -> str: +class _LineIteratorReader(io.RawIOBase): + """Adapter to convert line iterator to file-like object for XML parsing.""" + + def __init__(self, lines: Iterator[str]): + self._lines = lines + self._buffer = b'' + self._exhausted = False + + def readable(self) -> bool: + return True + + def readinto(self, b): # type: ignore[override] + """Read bytes from line iterator into buffer.""" + if self._exhausted and not self._buffer: + return 0 + + # Fill buffer from line iterator + while len(self._buffer) < len(b) and not self._exhausted: + try: + line = next(self._lines) + self._buffer += line.encode('utf-8') + except StopIteration: + self._exhausted = True + break + + # Copy buffer to output + n = min(len(self._buffer), len(b)) + b[:n] = self._buffer[:n] + self._buffer = self._buffer[n:] + return n + + +def _wrap_description_xml(description_xml: str) -> str: """Wrap a Description element in a minimal valid XML document. + ElementTree preserves namespace URIs when serializing elements, so we + only need to provide the minimal RDF namespace declaration. Additional + namespaces are already included in the serialized element. + Args: - xml_declaration: XML declaration line () - rdf_root: Root RDF element with namespaces () - description_lines: Lines making up the Description element + description_xml: Serialized Description element Returns: Complete XML document string """ - if xml_declaration is None: - xml_declaration = '' - - if rdf_root is None: - # Fallback minimal namespace declaration - rdf_root = '' - - # Ensure rdf_root is properly closed for wrapping - if not rdf_root.rstrip().endswith(">"): - rdf_root = rdf_root.rstrip() + ">" - - description_text = "".join(description_lines) - - # Construct minimal valid XML document - wrapped = f"{xml_declaration}\n{rdf_root}\n{description_text}\n" - + # Minimal RDF/XML wrapper - ElementTree preserves other namespaces + wrapped = ( + '\n' + '\n' + f'{description_xml}\n' + '' + ) return wrapped @@ -288,15 +314,13 @@ def stream_rdfxml_chunks_from_lines( lines: Iterator[str], chunk_size: int = DEFAULT_BATCH_SIZE, ) -> Iterator[list[dict]]: - """Stream RDF/XML format from a line iterator. + """Stream RDF/XML format from a line iterator using standard XML parser. - Strategy: - 1. Extract root element with namespaces (first 2 lines) - 2. Track state: inside/outside Description element - 3. Accumulate lines until is found - 4. Wrap each Description in minimal XML document with namespaces - 5. Parse each Description separately - 6. Extract all triples from each Description + Uses xml.etree.ElementTree.iterparse for robust XML parsing that handles: + - XML comments, CDATA, processing instructions + - Multi-line elements and attributes + - Nested elements with correct depth tracking + - All XML edge cases per W3C specification Args: lines: Iterator yielding text lines of RDF/XML data @@ -306,134 +330,81 @@ def stream_rdfxml_chunks_from_lines( Lists of triple dictionaries """ console = Console() - console.print("[yellow]Streaming RDF/XML from line iterator[/yellow]") + console.print("[yellow]Streaming RDF/XML with standard XML parser[/yellow]") + + # Convert line iterator to file-like object for XML parser + file_obj = io.BufferedReader(_LineIteratorReader(lines)) - # State tracking - root_line: str | None = None # XML declaration - rdf_root_line: str | None = None # with namespaces - current_description: list[str] = [] - in_description = False - description_count = 0 triple_batch: list[dict] = [] - line_count = 0 + batch_description_count = 0 + total_descriptions = 0 - # Track XML element depth for nested elements - depth = 0 + try: + # Use iterparse to process XML incrementally + # Get 'end' events when elements are complete + for _event, elem in ET.iterparse(file_obj, events=('end',)): + # Process complete Description elements + if _is_description_element(elem): + total_descriptions += 1 - for line in lines: - line_count += 1 - stripped = line.strip() + # Serialize element to string (preserves namespaces) + description_xml = ET.tostring(elem, encoding='unicode') - # Skip empty lines - if not stripped: - continue + # Wrap for rdflib parsing + wrapped_xml = _wrap_description_xml(description_xml) - # Extract XML declaration (line 1) - if stripped.startswith(" - if stripped.rstrip().endswith("/>"): - # Self-closing, process immediately - in_description = False - depth = 0 - description_xml = _wrap_description_xml( - root_line, rdf_root_line, current_description - ) - triples = _parse_description_xml(description_xml) - triple_batch.extend(triples) - description_count += 1 - current_description = [] - - if description_count >= chunk_size: + # Yield when chunk_size reached + if batch_description_count >= chunk_size and triple_batch: yield triple_batch triple_batch = [] - description_count = 0 - continue + batch_description_count = 0 - # If we're inside a Description, accumulate lines - if in_description: - current_description.append(line) - - # Count opening/closing tags to track depth correctly - # Opening tags: or (not ) - # Find all because regex requires [a-zA-Z_] after <) - all_tag_starts = re.findall(r'<([a-zA-Z_][\w:.-]*)', stripped) - # Closing tags: - closing_tags = len(re.findall(r'', stripped)) - # Self-closing tags: (these don't change depth) - # Find tags that end with /> before the next > - self_closing_tags = len(re.findall(r'<([a-zA-Z_][\w:.-]*)[^>]*/>', stripped)) - - # Opening tags = all tag starts - self-closing tags - # (all_tag_starts already excludes closing tags, so we only subtract self-closing) - opening_tags = len(all_tag_starts) - self_closing_tags - - # Net depth change: opening tags increase depth, closing tags decrease it - # Self-closing tags don't change depth (already excluded from opening_tags) - depth += opening_tags - closing_tags - - # Check if Description element is closed - if "" in stripped: - # The closing tag was already accounted for in closing_tags above - if depth == 0: - # Complete Description element found - in_description = False - - # Wrap in minimal XML document - description_xml = _wrap_description_xml( - root_line, rdf_root_line, current_description + # Progress logging every N descriptions + if total_descriptions % PROGRESS_LOG_INTERVAL == 0: + console.print( + f"[dim]Processed {total_descriptions:,} " + f"Description elements...[/dim]" ) - # Parse and extract triples - triples = _parse_description_xml(description_xml) - triple_batch.extend(triples) - description_count += 1 - - # Reset for next Description - current_description = [] - - # Yield when chunk_size reached - if description_count >= chunk_size: - yield triple_batch - triple_batch = [] - description_count = 0 - - # Handle other elements (like owl:Ontology) - skip for now - # Could be extended to parse these too if needed - - if line_count % PROGRESS_LOG_INTERVAL == 0: - console.print( - f"[dim]Processed {line_count:,} lines, {description_count:,} descriptions...[/dim]" - ) - - # Process final Description if file ends mid-element - if current_description and in_description: - # Try to parse anyway (might be incomplete) - try: - description_xml = _wrap_description_xml( - root_line, rdf_root_line, current_description - ) - triples = _parse_description_xml(description_xml) - triple_batch.extend(triples) - except Exception as e: - logger.warning(f"Incomplete Description at end of file: {e}") + except ET.ParseError as e: + logger.error(f"XML parsing error: {e}") + logger.error("This may indicate malformed XML in the source file") # Yield final batch if triple_batch: yield triple_batch + console.print( + f"[green]Completed: processed {total_descriptions:,} " + f"Description elements[/green]" + ) + + +def _is_description_element(elem: ET.Element) -> bool: + """Check if element is an rdf:Description element. + + Args: + elem: XML element + + Returns: + True if element is rdf:Description + """ + # Handle both namespaced and non-namespaced forms + tag = elem.tag + return ( + tag.endswith('Description') and + ('rdf' in tag.lower() or 'Description' in tag) + ) + def stream_rdf_chunks_from_lines( lines: Iterator[str], @@ -444,7 +415,8 @@ def stream_rdf_chunks_from_lines( Args: lines: Iterator yielding text lines of RDF data - format: RDF serialization format (supported: nt, ntriples, turtle, ttl, xml, rdf, rdfxml) + format: RDF serialization format + (supported: nt, ntriples, turtle, ttl, xml, rdf, rdfxml) chunk_size: Number of triples per chunk for Turtle parsing Yields: @@ -474,7 +446,7 @@ def stream_rdf_chunks_from_lines( def _pick_zip_member( zf: zipfile.ZipFile, *, - prefer_exts: tuple[str, ...] = (".ttl", ".nt", ".rdf", ".xml") + prefer_exts: tuple[str, ...] = (".ttl", ".nt", ".rdf", ".xml"), ) -> str: """Select the best RDF member file from a ZIP archive.""" candidates = [] @@ -531,7 +503,7 @@ def open_rdf_line_stream_from_url( lines = iter_text_lines( member_bytes, encoding=extract_cfg.encoding, - errors=extract_cfg.errors + errors=extract_cfg.errors, ) return lines, f"{name}:{member_name}" raise ValueError("No .ttl/.nt/.rdf/.xml member found in tar archive stream.") @@ -554,15 +526,11 @@ def open_rdf_line_stream_from_url( os.remove(tmp_path) # Compressed files and plain text - compression_format = detect_compression_format(name) - - decompressed = iter_decompressed_bytes( - byte_iter = byte_iter, - ) + decompressed = iter_decompressed_bytes(name, byte_iter) lines = iter_text_lines( decompressed, encoding=extract_cfg.encoding, - errors=extract_cfg.errors + errors=extract_cfg.errors, ) return lines, name @@ -627,22 +595,23 @@ def create_rdf_chunk_iterator( if input_type == "file": # Local file mode: use optimized file-based streaming with multiprocessing console.print("[cyan]Mode: Local file streaming[/cyan]") - console.print(f"[dim]Using parallel file-based parser ({num_workers} " - "workers)[/dim]") + console.print( + f"[dim]Using parallel file-based parser " + f"({num_workers} workers)[/dim]" + ) file_path = Path(input_value) chunk_iter = stream_rdf_chunks( - file_path, - rdf_format, - DEFAULT_BATCH_SIZE, - num_workers + file_path, rdf_format, DEFAULT_BATCH_SIZE, num_workers ) return chunk_iter, f"file:{file_path.name}" elif input_type == "registry": # Streaming download mode: use line-based streaming console.print("[cyan]Mode: Streaming download from registry[/cyan]") - console.print("[dim]Using line-based streaming parser (no " - "multiprocessing)[/dim]") + console.print( + "[dim]Using line-based streaming parser " + "(no multiprocessing)[/dim]" + ) ds = DATASET_REGISTRY[input_value] url = ds.url @@ -660,9 +629,7 @@ def create_rdf_chunk_iterator( console.print(f"[green]✓ Streaming source ready:[/green] {source_name}") chunk_iter = stream_rdf_chunks_from_lines( - lines, - format=rdf_format, - chunk_size=DEFAULT_BATCH_SIZE + lines, format=rdf_format, chunk_size=DEFAULT_BATCH_SIZE ) return chunk_iter, source_name @@ -767,9 +734,12 @@ def convert_and_upload_incremental( except Exception as e: error_str = str(e).lower() # Check if it's a rate limit error (429) - if ("429" in error_str - or "too many requests" in error_str - or "rate limit" in error_str): + is_rate_limit = ( + "429" in error_str + or "too many requests" in error_str + or "rate limit" in error_str + ) + if is_rate_limit: if attempt < max_retries - 1: # Exponential backoff with jitter delay = min(base_delay * (2 ** attempt), max_delay) @@ -784,8 +754,10 @@ def convert_and_upload_incremental( time.sleep(wait_time) buffer.seek(0) # Reset buffer position for retry else: - console.print("[red]✗ Max retries reached for shard " - f"{current_shard_idx}[/red]") + console.print( + f"[red]✗ Max retries reached for " + f"shard {current_shard_idx}[/red]" + ) raise else: # Not a rate limit error, re-raise immediately @@ -837,8 +809,10 @@ def convert_and_upload_incremental( ensure_writer() finalize_and_upload(shard_idx, shard_rows) - console.print("[green]✓ Uploaded shard " - f"{shard_idx} ({shard_rows:,} rows)[/green]") + console.print( + f"[green]✓ Uploaded shard {shard_idx} " + f"({shard_rows:,} rows)[/green]" + ) # Add small delay to avoid overwhelming HuggingFace API time.sleep(1.0) @@ -846,8 +820,10 @@ def convert_and_upload_incremental( shard_rows = 0 if max_shards is not None and shard_idx >= max_shards: - console.print("[yellow]Stopping after " - f"max_shards={max_shards}[/yellow]") + console.print( + f"[yellow]Stopping after " + f"max_shards={max_shards}[/yellow]" + ) break if max_shards is not None and shard_idx >= max_shards: @@ -863,8 +839,10 @@ def convert_and_upload_incremental( ensure_writer() finalize_and_upload(shard_idx, shard_rows) - console.print("[green]✓ Uploaded final shard " - f"{shard_idx} ({shard_rows:,} rows)[/green]") + console.print( + f"[green]✓ Uploaded final shard {shard_idx} " + f"({shard_rows:,} rows)[/green]" + ) shard_idx += 1 elapsed = time.time() - start_time @@ -915,11 +893,15 @@ def list_datasets() -> None: for info in DATASET_REGISTRY.values(): by_category[info.category] = by_category.get(info.category, 0) + 1 - console.print(f"\n[bold]Total:[/bold] {total} datasets " - f"({available_count} available)") + console.print( + f"\n[bold]Total:[/bold] {total} datasets " + f"({available_count} available)" + ) console.print("[bold]By category:[/bold] ", end="") - console.print(", ".join(f"{cat}: {count}" - for cat, count in sorted(by_category.items()))) + cat_list = ", ".join( + f"{cat}: {count}" for cat, count in sorted(by_category.items()) + ) + console.print(cat_list) console.print() @@ -1008,14 +990,16 @@ size_categories: ### Dataset Summary -This dataset contains RDF triples from {dataset_info.name} converted to HuggingFace -dataset format for easy use in machine learning pipelines. +This dataset contains RDF triples from {dataset_info.name} converted to +HuggingFace dataset format for easy use in machine learning pipelines. -- **Format:** Originally {dataset_info.format}, converted to HuggingFace Dataset +- **Format:** Originally {dataset_info.format}, converted to HuggingFace + Dataset - **Size:** {dataset_info.size_gb} GB (extracted) - **Entities:** {dataset_info.entities if dataset_info.entities else "N/A"} - **Triples:** {dataset_info.triples if dataset_info.triples else "N/A"} -- **Original License:** {dataset_info.license if dataset_info.license else "See original source"} +- **Original License:** {dataset_info.license if dataset_info.license else + "See original source"} ### Recommended Use @@ -1025,7 +1009,8 @@ dataset format for easy use in machine learning pipelines. ## RDF Format -This dataset uses a standard lossless format for representing RDF triples. Each triple is a row with 6 fields: +This dataset uses a standard lossless format for representing RDF triples. +Each triple is a row with 6 fields: - `subject`: Subject URI or blank node - `predicate`: Predicate URI - `object`: Object value (URI, literal, or blank node) @@ -1071,7 +1056,10 @@ This dataset is part of the CleverThis knowledge graph collection. def main() -> int: """Unified RDF to HuggingFace converter with automatic mode detection.""" parser = argparse.ArgumentParser( - description="Convert RDF to HuggingFace with incremental Parquet writing", + description=( + "Convert RDF to HuggingFace dataset " + "with incremental Parquet writing" + ), formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Input Modes: @@ -1091,7 +1079,8 @@ Examples: python rdf_to_hf_incremental.py /data/wordnet.nt username/wordnet-dataset # Process local file with explicit format - python rdf_to_hf_incremental.py /data/wordnet.nt username/wordnet-dataset --format ntriples + python rdf_to_hf_incremental.py /data/wordnet.nt username/wordnet-dataset \\ + --format ntriples # Production with checkpointing python rdf_to_hf_incremental.py dbpedia-core username/dbpedia \\ @@ -1175,7 +1164,10 @@ Features: "--num-workers", type=int, default=None, - help="Number of processes for local file parsing (default: CPU count - 1)", + help=( + "Number of worker processes for local file parsing " + "(default: CPU count - 1)" + ), ) parser.add_argument( "--verify-after", @@ -1220,10 +1212,14 @@ Features: # Validate required arguments if not in list mode if not args.input or not args.repo_id: - console.print("[red]Error: input and repo_id are required " - "(unless using --list)[/red]") - console.print("[yellow]Usage: rdf_to_hf_incremental.py INPUT REPO_ID " - "[OPTIONS][/yellow]") + console.print( + "[red]Error: input and repo_id are required " + "(unless using --list)[/red]" + ) + console.print( + "[yellow]Usage: rdf_to_hf_incremental.py " + "INPUT REPO_ID [OPTIONS][/yellow]" + ) console.print("[yellow] or: rdf_to_hf_incremental.py --list[/yellow]") return 1 @@ -1281,25 +1277,31 @@ Features: ".trig": "trig" } rdf_format = format_map.get(ext, "turtle") - console.print(f"[dim]Format inferred from extension: {ext} → {rdf_format}[/dim]") + console.print( + f"[dim]Format inferred from extension: " + f"{ext} → {rdf_format}[/dim]" + ) - console.print(f"[dim]Rows per shard: {args.rows_per_shard:,}, " - f"Record batch size: {args.record_batch_size:,}[/dim]") + console.print( + f"[dim]Rows per shard: {args.rows_per_shard:,}, " + f"Record batch size: {args.record_batch_size:,}[/dim]" + ) # Handle --dry-run mode if args.dry_run: console.print("\n[bold yellow]═══ DRY RUN MODE ═══[/bold yellow]") - console.print("[yellow]Showing what would be done:[/yellow]\n") + console.print( + "[yellow]Showing what would be done without executing:[/yellow]\n" + ) console.print(f"[cyan]Input type:[/cyan] {input_type}") console.print(f"[cyan]Input value:[/cyan] {input_value}") console.print(f"[cyan]RDF format:[/cyan] {rdf_format}") console.print(f"[cyan]Target repo:[/cyan] {args.repo_id}") console.print(f"[cyan]Rows per shard:[/cyan] {args.rows_per_shard:,}") - console.print("[cyan]Record batch size:[/cyan] " - f"{args.record_batch_size:,}") - console.print("[cyan]Num workers:[/cyan] " - f"{args.num_workers or 'auto (CPU count - 1)'}") + console.print(f"[cyan]Record batch size:[/cyan] {args.record_batch_size:,}") + num_workers_display = args.num_workers or 'auto (CPU count - 1)' + console.print(f"[cyan]Num workers:[/cyan] {num_workers_display}") console.print(f"[cyan]Checkpoint file:[/cyan] {args.checkpoint or 'None'}") if input_type == "registry": @@ -1314,21 +1316,33 @@ Features: # Handle both string and numeric triple counts if isinstance(ds.triples, (int, float)): console.print(f" Estimated triples: {int(ds.triples):,}") - estimated_shards = (int(ds.triples) // args.rows_per_shard) + 1 - console.print(f" Estimated shards: ~{estimated_shards:,}") + estimated_shards = ( + (int(ds.triples) // args.rows_per_shard) + 1 + ) + console.print( + f" Estimated shards: ~{estimated_shards:,}" + ) else: console.print(f" Estimated triples: {ds.triples}") 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}") # noqa: E501 + if input_type == 'registry': + action_1 = f"Stream from {DATASET_REGISTRY[input_value].url}" + else: + action_1 = f"Read from {input_value}" + console.print(f" 1. {action_1}") console.print(f" 2. Parse RDF triples in {rdf_format} format") - console.print(" 3. Convert to Parquet shards " - f"({args.rows_per_shard:,} rows each)") + console.print( + f" 3. Convert to Parquet shards " + f"({args.rows_per_shard:,} rows each)" + ) console.print(f" 4. Upload shards to {args.repo_id}") console.print(" 5. Create and upload README.md") - console.print("\n[bold green]✓ Dry run complete. " - "Use without --dry-run to execute.[/bold green]") + console.print( + "\n[bold green]✓ Dry run complete. " + "Use without --dry-run to execute.[/bold green]" + ) return 0 # Initialize HuggingFace API @@ -1336,8 +1350,9 @@ Features: if token: console.print("[dim]Using provided/environment token[/dim]") else: - console.print("[dim]Using cached credentials from " - "'huggingface-cli login'[/dim]") + console.print( + "[dim]Using cached credentials from 'huggingface-cli login'[/dim]" + ) try: api = HfApi(token=token) @@ -1393,11 +1408,16 @@ Features: start_shard=start_shard, ) - console.print("\n[bold green]✓ Successfully uploaded to " - f"{args.repo_id}[/bold green]") + console.print( + f"\n[bold green]✓ Successfully uploaded to " + f"{args.repo_id}[/bold green]" + ) console.print(f"[green] Total rows: {total_rows:,}[/green]") console.print(f"[green] Total shards: {total_shards}[/green]") - console.print(f"[green] Repository: https://huggingface.co/datasets/{args.repo_id}[/green]") + console.print( + f"[green] Repository: " + f"https://huggingface.co/datasets/{args.repo_id}[/green]" + ) # Create and upload README.md (dataset card) if input_type == "registry": @@ -1419,35 +1439,41 @@ Features: console.print("[green]✓ Dataset card (README.md) " "uploaded[/green]") except Exception as e: - console.print("[yellow]⚠ Warning: Could not upload README.md: " - f"{e}[/yellow]") - console.print("[dim]You can manually create a README.md file " - "for the dataset[/dim]") + console.print( + f"[yellow]⚠ Warning: Could not upload README.md: " + f"{e}[/yellow]" + ) + console.print( + "[dim]You can manually create a README.md file " + "for the dataset[/dim]" + ) # Verification if args.verify_after: console.print("\n[yellow]Verifying uploaded shards...[/yellow]") try: repo_files = api.list_repo_files( - repo_id=args.repo_id, - repo_type="dataset" + repo_id=args.repo_id, repo_type="dataset" ) data_files = [f for f in repo_files if f.startswith("data/")] - console.print(f"[green]✓ Found {len(data_files)} shard files " - "in repo[/green]") + console.print( + f"[green]✓ Found {len(data_files)} shard files " + f"in repo[/green]" + ) sample_n = min(args.verify_sample_shards, len(data_files)) for f in data_files[:sample_n]: local_path = hf_hub_download( repo_id=args.repo_id, repo_type="dataset", - filename=f + filename=f, ) pf = pq.ParquetFile(local_path) console.print(f"[green]✓ Verified readable:[/green] {f}") - console.print( - "[dim] Rows: " - f"{pf.metadata.num_rows if pf.metadata else 'unknown'}[/dim]") + num_rows = ( + pf.metadata.num_rows if pf.metadata else 'unknown' + ) + console.print(f"[dim] Rows: {num_rows}[/dim]") except Exception as e: console.print(f"[red]✗ Verification failed: {e}[/red]") diff --git a/src/streaming/http_stream.py b/src/streaming/http_stream.py index a65254d..90f2c27 100644 --- a/src/streaming/http_stream.py +++ b/src/streaming/http_stream.py @@ -21,6 +21,13 @@ class HTTPStreamConfig: follow_redirects: bool = True +def _calculate_backoff_sleep( + attempt: int, base_s: float, max_s: float +) -> float: + """Calculate exponential backoff sleep time.""" + return min(base_s * (2 ** (attempt - 1)), max_s) + + def stream_http_bytes( url: str, *, @@ -41,9 +48,14 @@ def stream_http_bytes( if protocol == "ftp": yield from stream_ftp_bytes(url, config=config) elif protocol in ("http", "https"): - yield from stream_http_bytes_impl(url, headers=headers, start_byte=start_byte, config=config) + yield from stream_http_bytes_impl( + url, headers=headers, start_byte=start_byte, config=config + ) else: - raise ValueError(f"Unsupported protocol: {protocol}. Supported: http, https, ftp") + raise ValueError( + f"Unsupported protocol: {protocol}. " + f"Supported: http, https, ftp" + ) def stream_http_bytes_impl( @@ -85,10 +97,11 @@ def stream_http_bytes_impl( yield chunk return except ( - httpx.TimeoutException, - httpx.NetworkError, - httpx.RemoteProtocolError, - httpx.HTTPStatusError) as e: + httpx.TimeoutException, + httpx.NetworkError, + httpx.RemoteProtocolError, + httpx.HTTPStatusError, + ) as e: # Retry on network/timeouts and 5xx. For 4xx (except 429), fail fast. if isinstance(e, httpx.HTTPStatusError): status = e.response.status_code @@ -98,10 +111,12 @@ def stream_http_bytes_impl( if status < 500 and status != 429: raise - if attempt > cfg.max_retries: + if attempt >= cfg.max_retries: raise - sleep_s = min(cfg.backoff_base_s * (2 ** (attempt - 1)), cfg.backoff_max_s) + sleep_s = _calculate_backoff_sleep( + attempt, cfg.backoff_base_s, cfg.backoff_max_s + ) time.sleep(sleep_s) @@ -152,22 +167,28 @@ def stream_ftp_bytes( raise ValueError(f"File not found on FTP server: {url}") from e elif "timed out" in reason.lower(): # Retry on timeout - if attempt > cfg.max_retries: + if attempt >= cfg.max_retries: raise - sleep_s = min(cfg.backoff_base_s * (2 ** (attempt - 1)), cfg.backoff_max_s) + sleep_s = _calculate_backoff_sleep( + attempt, cfg.backoff_base_s, cfg.backoff_max_s + ) time.sleep(sleep_s) else: # Other errors: retry if we have attempts left - if attempt > cfg.max_retries: + if attempt >= cfg.max_retries: raise - sleep_s = min(cfg.backoff_base_s * (2 ** (attempt - 1)), cfg.backoff_max_s) + sleep_s = _calculate_backoff_sleep( + attempt, cfg.backoff_base_s, cfg.backoff_max_s + ) time.sleep(sleep_s) - except Exception as e: + except Exception: # Unexpected errors: retry if we have attempts left - if attempt > cfg.max_retries: + if attempt >= cfg.max_retries: raise - sleep_s = min(cfg.backoff_base_s * (2 ** (attempt - 1)), cfg.backoff_max_s) + sleep_s = _calculate_backoff_sleep( + attempt, cfg.backoff_base_s, cfg.backoff_max_s + ) time.sleep(sleep_s) -- 2.52.0 From 255ccad4d4a1c70bd5e2f4eb2f72935b9fe81304 Mon Sep 17 00:00:00 2001 From: Aditya Chhabra Date: Tue, 6 Jan 2026 19:56:41 +0530 Subject: [PATCH 3/4] fix: some minor fixes --- scripts/rdf_to_hf_incremental.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/scripts/rdf_to_hf_incremental.py b/scripts/rdf_to_hf_incremental.py index f699a11..f12c25d 100644 --- a/scripts/rdf_to_hf_incremental.py +++ b/scripts/rdf_to_hf_incremental.py @@ -142,12 +142,10 @@ def stream_turtle_chunks_from_lines( current_chunk: list[str] = [] triple_count = 0 - line_count = 0 prefix_lines: list[str] = [] in_prefixes = True - for line in lines: - line_count += 1 + for line_count, line in enumerate(lines, start=1): stripped = line.strip() if not stripped or stripped.startswith("#"): -- 2.52.0 From b696d8683594b96a82c964e701ac29422c5a4415 Mon Sep 17 00:00:00 2001 From: Aditya Chhabra Date: Wed, 14 Jan 2026 18:12:27 +0530 Subject: [PATCH 4/4] fix: fix minor decompressed_byte bug --- scripts/rdf_to_hf_incremental.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/scripts/rdf_to_hf_incremental.py b/scripts/rdf_to_hf_incremental.py index f12c25d..86aacda 100644 --- a/scripts/rdf_to_hf_incremental.py +++ b/scripts/rdf_to_hf_incremental.py @@ -47,6 +47,8 @@ from xml.etree import ElementTree as ET import pyarrow as pa import pyarrow.parquet as pq +from convert_rdf_to_hf_dataset_streaming_parallel import stream_rdf_chunks +from dataset_registry import DATASET_REGISTRY, DatasetInfo, get_dataset_config from huggingface_hub import HfApi, hf_hub_download from rdflib import Graph, Literal, URIRef from rich.console import Console @@ -496,7 +498,7 @@ def open_rdf_line_stream_from_url( # TAR archives if lower.endswith((".tar.gz", ".tgz", ".tar")): - for member_name, member_bytes in iter_tar_member_bytes(name, byte_iter): + for member_name, member_bytes in iter_tar_member_bytes(byte_iter, tar_name=name): if member_name.lower().endswith((".ttl", ".nt", ".rdf", ".xml")): lines = iter_text_lines( member_bytes, @@ -524,7 +526,7 @@ def open_rdf_line_stream_from_url( os.remove(tmp_path) # Compressed files and plain text - decompressed = iter_decompressed_bytes(name, byte_iter) + decompressed = iter_decompressed_bytes(byte_iter, compressed_name=name) lines = iter_text_lines( decompressed, encoding=extract_cfg.encoding, -- 2.52.0