# Data Collectors ## I. System Architecture ``` ┌─────────────────────────────────────────────────────────────────┐ │ Data Collection Architecture │ ├─────────────────────────────────────────────────────────────────┤ │ │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ TOP500 │ │ Epoch AI │ │ HuggingFace │ │ │ │ Collector │ │ Collector │ │ Collector │ │ │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │ │ │ │ │ │ │ └───────────────────┼───────────────────┘ │ │ ▼ │ │ ┌─────────────────────┐ │ │ │ BaseCollector │◄── Base class (unified) │ │ │ run() method │ │ │ └─────────┬───────────┘ │ │ │ │ │ ┌─────────────────┼─────────────────┐ │ │ ▼ ▼ ▼ │ │ ┌───────────┐ ┌───────────┐ ┌───────────┐ │ │ │ fetch() │ │transform()│ │ _save_data│ │ │ │ raw data │ │ transform │ │ save to DB│ │ │ └───────────┘ └───────────┘ └───────────┘ │ │ │ │ │ ▼ │ │ ┌─────────────────────┐ │ │ │ CollectedData table│◄── Unified storage │ │ └─────────────────────┘ │ │ │ │ ┌─────────────────────────────────────────────────────────┐ │ │ │ Scheduler (APScheduler) │ │ │ │ Scheduled tasks: every 4h/6h/12h/1d auto-execute │ │ │ └─────────────────────────────────────────────────────────┘ │ │ │ └─────────────────────────────────────────────────────────────────┘ ``` ## II. Pipeline ```python # 1. Scheduler triggers (scheduled or manual) # ↓ # 2. run() executes the full pipeline async def run(self, db): # 2.1 Check if collector is enabled if not collector_registry.is_active(self.name): return {"status": "skipped"} # 2.2 Record task start task = CollectionTask(status="running") db.add(task) await db.commit() # 2.3 FETCH — get raw data (implemented by subclass) raw_data = await self.fetch() # 2.4 TRANSFORM — convert to unified format data = self.transform(raw_data) # 2.5 SAVE — persist to database records_count = await self._save_data(db, data) # 2.6 Record task completion task.status = "success" task.records_processed = records_count await db.commit() ``` **Core file**: `backend/app/services/collectors/base.py` ## III. Collector List | Collector | Data type | Content | Frequency | |-----------|-----------|---------|-----------| | TOP500 | supercomputer | Global supercomputer rankings (compute, performance) | 4 hours | | Epoch AI | gpu_cluster | GPU compute cluster info | 6 hours | | HuggingFace Models | model | AI model information | 12 hours | | HuggingFace Datasets | dataset | Dataset information | 12 hours | | HuggingFace Spaces | space | Demo applications | 1 day | | PeeringDB | ixp/network/facility | Internet exchange points / networks / facilities | 1-2 days | | TeleGeography | submarine_cable | Submarine cable information | 7 days | ## IV. Data Format (stored in CollectedData table) ```python # Each collector's parse_response() return format { "source_id": "top500_1", # Original system ID (required) "name": "El Capitan", # Name (required) "description": "System desc...", # Description "country": "United States", # Country "city": "Livermore, CA", # City "latitude": "37.6819", # Latitude (string) "longitude": "-121.7681", # Longitude (string) "value": "1742.00", # Performance value (e.g. compute) "unit": "PFlop/s", # Unit "metadata": { # Extra data (JSON) "rank": 1, "r_peak": 2746.38, "cores": 11039616 }, "reference_date": "2025-11-01" # Data reference date } ``` ## V. Database Schema **CollectedData table** (`collected_data`) | Field | Type | Description | |-------|------|-------------| | id | SERIAL | Primary key | | source | VARCHAR(100) | Data source name (top500, huggingface, etc.) | | source_id | VARCHAR(100) | Original data ID | | data_type | VARCHAR(50) | Data type (supercomputer, model, etc.) | | name | VARCHAR(500) | Name | | title | VARCHAR(500) | Title | | description | TEXT | Description | | country | VARCHAR(100) | Country | | city | VARCHAR(100) | City | | latitude | VARCHAR(50) | Latitude | | longitude | VARCHAR(50) | Longitude | | value | VARCHAR(100) | Performance value | | unit | VARCHAR(20) | Unit | | metadata | JSONB | Extra metadata | | collected_at | TIMESTAMP | Collection time | | reference_date | TIMESTAMP | Data reference date | | is_valid | INTEGER | Whether valid | **Core file**: `backend/app/models/collected_data.py` ## VI. TOP500 Collector Example (full pipeline) ```python # 1. fetch() — get HTML from the web async def fetch(self): url = "https://top500.org/lists/top500/list/2025/11/" response = await client.get(url) return response.text # returns HTML # 2. parse_response() — parse HTML into unified format def parse_response(self, html): soup = BeautifulSoup(html, "html.parser") table = soup.find("table") for row in table.find_all("tr")[1:]: # skip header cells = row.find_all("td") entry = { "source_id": f"top500_{cells[0].text}", "name": cells[1].text.strip(), "country": cells[2].text.strip(), "city": "", "latitude": "", "longitude": "", "value": "1742.00", "unit": "PFlop/s", "metadata": { "rank": 1, "cores": "11340000" }, "reference_date": "2025-11-01" } data.append(entry) return data # 3. run() automatically calls _save_data() to save to database ``` **Core file**: `backend/app/services/collectors/top500.py` ## VII. Scheduler ```python # Register all collectors into scheduled tasks at startup def start_scheduler(): for name, collector in collectors.items(): if collector_registry.is_active(name): scheduler.add_job( run_collector_task, trigger=IntervalTrigger(hours=collector.frequency_hours), id=name, name=name ) ``` | Collector | Frequency | |-----------|-----------| | TOP500 | Every 4 hours | | Epoch AI | Every 6 hours | | HuggingFace | Every 12 hours | | PeeringDB | Every 1-2 days | | TeleGeography | Every 7 days | **Core file**: `backend/app/services/scheduler.py` ## VIII. Code Files ``` backend/app/services/collectors/ ├── base.py # Base class: run() pipeline, _save_data() persistence ├── registry.py # Collector registry ├── scheduler.py # Scheduled task dispatch (APScheduler) ├── top500.py # TOP500 collector ├── epoch_ai.py # Epoch AI collector ├── huggingface.py # HuggingFace collector ├── peeringdb.py # PeeringDB collector └── telegeraphy.py # TeleGeography submarine cable collector backend/app/models/ └── collected_data.py # Unified data model ``` ## IX. Data Usage Collected data ultimately: 1. **Visualization** — displays supercomputers, GPU clusters, and submarine cables' geographic positions 2. **Situational analysis** — global compute distribution statistics and growth trends 3. **Alert system** — detects changes to important nodes ## X. Collector Registration Collectors are automatically registered at application startup: ```python # backend/app/services/collectors/__init__.py collector_registry.register(TOP500Collector()) collector_registry.register(EpochAIGPUCollector()) collector_registry.register(HuggingFaceModelCollector()) collector_registry.register(HuggingFaceDatasetCollector()) collector_registry.register(HuggingFaceSpacesCollector()) collector_registry.register(PeeringDBIXPCollector()) collector_registry.register(PeeringDBNetworkCollector()) collector_registry.register(PeeringDBFacilityCollector()) collector_registry.register(TeleGeographyCableCollector()) collector_registry.register(TeleGeographyLandingPointCollector()) collector_registry.register(TeleGeographyCableSystemCollector()) ``` **Core file**: `backend/app/services/collectors/registry.py` ## XI. Triggering Collection ### Method 1: Scheduled At startup, APScheduler automatically creates scheduled tasks based on each collector's `frequency_hours` setting. ### Method 2: Manual API trigger ```bash # Trigger TOP500 collection curl -X POST http://localhost:8000/api/v1/datasources/1/trigger \ -H "Authorization: Bearer " ``` **Core file**: `backend/app/api/v1/datasources.py`