14 KiB
14 KiB
数据采集系统 (Collectors)
一、系统架构
┌─────────────────────────────────────────────────────────────────┐
│ 数据采集系统架构 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ TOP500 │ │ Epoch AI │ │ HuggingFace │ │
│ │ 采集器 │ │ 采集器 │ │ 采集器 │ │
│ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │
│ │ │ │ │
│ └───────────────────┼───────────────────┘ │
│ ▼ │
│ ┌─────────────────────┐ │
│ │ BaseCollector │◄── 基类 (统一处理) │
│ │ run() 方法 │ │
│ └─────────┬───────────┘ │
│ │ │
│ ┌─────────────────┼─────────────────┐ │
│ ▼ ▼ ▼ │
│ ┌───────────┐ ┌───────────┐ ┌───────────┐ │
│ │ fetch() │ │transform()│ │ _save_data│ │
│ │ 获取原始数据 │ │ 数据转换 │ │ 保存到DB │ │
│ └───────────┘ └───────────┘ └───────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────┐ │
│ │ CollectedData 表 │◄── 统一存储 │
│ └─────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ Scheduler (APScheduler) │ │
│ │ 定时任务调度: 每4小时/6小时/12小时/1天 自动执行 │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
二、工作流程 (Pipeline)
# 1. Scheduler 触发 (定时 或 手动触发)
# ↓
# 2. run() 方法执行完整流水线
async def run(self, db):
# 2.1 检查采集器是否启用
if not collector_registry.is_active(self.name):
return {"status": "skipped"}
# 2.2 记录任务开始
task = CollectionTask(status="running")
db.add(task)
await db.commit()
# 2.3 FETCH - 获取原始数据 (由子类实现)
raw_data = await self.fetch()
# 2.4 TRANSFORM - 转换为统一格式
data = self.transform(raw_data)
# 2.5 SAVE - 保存到数据库
records_count = await self._save_data(db, data)
# 2.6 记录任务完成
task.status = "success"
task.records_processed = records_count
await db.commit()
核心文件: backend/app/services/collectors/base.py
三、采集器列表
| 采集器 | 数据类型 | 数据内容 | 采集频率 |
|---|---|---|---|
| TOP500 | supercomputer | 全球超级计算机排名 (算力、性能) | 4小时 |
| Epoch AI | gpu_cluster | GPU算力集群信息 | 6小时 |
| HuggingFace Models | model | AI模型信息 | 12小时 |
| HuggingFace Datasets | dataset | 数据集信息 | 12小时 |
| HuggingFace Spaces | space | Demo应用 | 1天 |
| PeeringDB | ixp/network/facility | 互联网交换点/网络/机房 | 1-2天 |
| TeleGeography | submarine_cable | 海底光缆信息 | 7天 |
| Space-Track TLE | satellite_tle | 卫星轨道 TLE 数据 | 依采集器配置 |
| BarentsWatch AIS | vessel | 船只位置、航速、航向、MMSI 等 AIS 数据 | 依采集器配置 |
四、数据格式 (统一存储到 CollectedData 表)
# 每个采集器 parse_response() 返回格式
{
"source_id": "top500_1", # 原始系统ID (必填)
"name": "El Capitan", # 名称 (必填)
"description": "系统描述...", # 描述
"country": "United States", # 国家
"city": "Livermore, CA", # 城市
"latitude": "37.6819", # 纬度 (字符串)
"longitude": "-121.7681", # 经度 (字符串)
"value": "1742.00", # 性能值 (如算力)
"unit": "PFlop/s", # 单位
"metadata": { # 额外数据 (JSON)
"rank": 1,
"r_peak": 2746.38,
"cores": 11039616
},
"reference_date": "2025-11-01" # 数据参考日期
}
五、数据库表结构
CollectedData 表 (collected_data)
| 字段 | 类型 | 说明 |
|---|---|---|
| id | SERIAL | 主键 |
| source | VARCHAR(100) | 数据源名称 (top500, huggingface等) |
| source_id | VARCHAR(100) | 原始数据ID |
| data_type | VARCHAR(50) | 数据类型 (supercomputer, model等) |
| name | VARCHAR(500) | 名称 |
| title | VARCHAR(500) | 标题 |
| description | TEXT | 描述 |
| country | VARCHAR(100) | 国家 |
| city | VARCHAR(100) | 城市 |
| latitude | VARCHAR(50) | 纬度 |
| longitude | VARCHAR(50) | 经度 |
| value | VARCHAR(100) | 性能值 |
| unit | VARCHAR(20) | 单位 |
| metadata | JSONB | 额外元数据 |
| collected_at | TIMESTAMP | 采集时间 |
| reference_date | TIMESTAMP | 数据参考日期 |
| is_valid | INTEGER | 是否有效 |
核心文件: backend/app/models/collected_data.py
六、TOP500 采集器示例 (完整流程)
# 1. fetch() - 从网页获取HTML
async def fetch(self):
url = "https://top500.org/lists/top500/list/2025/11/"
response = await client.get(url)
return response.text # 返回HTML
# 2. parse_response() - 解析HTML为统一格式
def parse_response(self, html):
soup = BeautifulSoup(html, "html.parser")
table = soup.find("table")
for row in table.find_all("tr")[1:]: # 跳过表头
cells = row.find_all("td")
entry = {
"source_id": f"top500_{cells[0].text}", # "top500_1"
"name": cells[1].text.strip(), # "El Capitan"
"country": cells[2].text.strip(), # "United States"
"city": "", # 城市
"latitude": "", # 需进一步解析
"longitude": "",
"value": "1742.00", # Rmax
"unit": "PFlop/s",
"metadata": {
"rank": 1,
"cores": "11340000"
},
"reference_date": "2025-11-01"
}
data.append(entry)
return data
# 3. run() 自动调用 _save_data() 保存到数据库
核心文件: backend/app/services/collectors/top500.py
七、调度机制
# 启动时注册所有采集器到定时任务
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
)
| 采集器 | 采集频率 |
|---|---|
| TOP500 | 每4小时 |
| Epoch AI | 每6小时 |
| HuggingFace | 每12小时 |
| PeeringDB | 每1-2天 |
| TeleGeography | 每7天 |
核心文件: backend/app/services/scheduler.py
成功采集与连接状态
内置采集器成功采集后，调度器会记录当前有效配置已通过连接验证：
if datasource.last_status == "success":
effective_candidate = await get_builtin_effective_candidate(db, datasource.source)
checksum, _ = await build_builtin_connectivity_checksum(...)
await save_connectivity_success(
db,
datasource.source,
checksum,
{"status_code": None},
connected_by="collection",
)
这个记录用于控制台“采集器设置”中的连接状态判断：如果当前配置和成功采集时的 checksum 一致，就视为已连接，不要求用户再手动点击连接按钮。只有 endpoint、请求头、基础配置或凭证指纹变化时，才需要重新验证。
相关实现见：
八、相关代码文件
backend/app/services/collectors/
├── base.py # 基类: run() 流水线, _save_data() 保存
├── registry.py # 采集器注册表
├── scheduler.py # 定时任务调度 (APScheduler)
├── top500.py # TOP500采集器
├── epoch_ai.py # Epoch AI采集器
├── huggingface.py # HuggingFace采集器
├── peeringdb.py # PeeringDB采集器
├── telegeraphy.py # TeleGeography海底光缆采集器
└── vessel_ais.py # BarentsWatch AIS 船只采集器
backend/app/models/
└── collected_data.py # 统一数据模型
九、凭证型采集器
部分采集器需要外部服务凭证，例如：
| 采集器 | credential provider | 凭证来源 |
|---|---|---|
barentswatch_vessels |
barentswatch |
控制台采集器设置、环境变量、~/.zshrc |
spacetrack_tle |
spacetrack |
环境变量、~/.zshrc |
BarentsWatch AIS
BarentsWatch AIS 的凭证解析统一在：
VesselAISCollector 只负责采集和转换 AIS 数据，不再自己读取环境变量或拼 token 请求。它通过：
resolve_barentswatch_config()fetch_barentswatch_access_token()
获取运行时配置。
解析优先级：
DataSourceConfig.auth_configDataSourceConfig.config- 环境变量
~/.zshrc
支持变量：
export BARENTSWATCH_CLIENT_ID="..."
export BARENTSWATCH_CLIENT_SECRET="..."
并兼容历史拼写：
export BARRENTSWATCH_CLIENT_ID="..."
export BARRENTSWATCH_CLIENT_SECRET="..."
连接验证会先请求 https://id.barentswatch.no/connect/token 获取 scope=ais 的 access token，再用 Authorization: Bearer <token> 请求 AIS endpoint。
十、采集器设置与连接验证
控制台的“采集器设置”页提供所有内置采集器的 endpoint、请求头、超时、重试和凭证配置。连接验证不是只看前端按钮状态，而是由后端计算 checksum：
- endpoint
- auth type
- headers
- config
- credential provider
- 凭证指纹
相关 API：
GET /api/v1/datasources/configs/all
POST /api/v1/datasources/configs/builtin/connection-status
POST /api/v1/datasources/configs/builtin/connect
POST /api/v1/settings/integrations/barentswatch/connect
GET /api/v1/settings/credential-guides/{provider}
POST /api/v1/settings/credential-guides/{provider}/generate
POST /api/v1/settings/credential-guides/{provider}/reset
更多细节见：
十一、数据使用场景
采集的数据最终会：
- 可视化展示 - 在UE5大屏上显示超级计算机、GPU集群、海底光缆的地理位置
- 态势分析 - 统计全球算力分布、增长趋势
- 告警系统 - 检测重要节点变化
十二、采集器注册机制
采集器在应用启动时自动注册：
# 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())
核心文件: backend/app/services/collectors/registry.py
十三、触发采集
方式一：定时触发
系统启动时，APScheduler会自动根据各采集器的frequency_hours设置定时任务。
方式二：手动触发 API
# 触发TOP500采集
curl -X POST http://localhost:8000/api/v1/datasources/1/trigger \
-H "Authorization: Bearer <token>"
核心文件: backend/app/api/v1/datasources.py