release: bump version to 0.42.0
This commit is contained in:
rayd1o
264
docs/technical/en/backend-collectors.md
Normal file
264
docs/technical/en/backend-collectors.md
Normal file
@@ -0,0 +1,264 @@
|
||||
# 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`
|
||||
Reference in New Issue
Block a user