265 lines
11 KiB
Markdown
265 lines
11 KiB
Markdown
# 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 <token>"
|
|
```
|
|
|
|
**Core file**: `backend/app/api/v1/datasources.py`
|