Analytics Engine 设计模式

# Analytics Engine 模式 ## 使用场景 | 使用场景 | 关键指标 | 索引字段 | |----------|-------------|----------| | API 计量 | requests, bytes, compute_units | api_key | | 功能使用 | feature, action, duration | user_id | | 错误追踪 | error_type, endpoint, count | customer_id | | 性能监控 | latency_ms, cache_status | endpoint | | A/B 测试 | variant, conversions | user_id | ## API 计量 (计费) typescript env.ANALYTICS.writeDataPoint({ blobs: [pathname, method, status, tier], doubles: [1, computeUnits, bytes, latencyMs], indexes: [apiKey] }); // 查询：按客户统计月度用量 // SELECT index1 AS api_key, SUM(double2) AS compute_units // FROM usage WHERE timestamp >= DATE_TRUNC('month', NOW()) GROUP BY index1 ## 错误追踪 typescript env.ANALYTICS.writeDataPoint({ blobs: [endpoint, method, errorName, errorMessage.slice(0, 1000)], doubles: [1, timeToErrorMs], indexes: [customerId] }); ## 性能监控 typescript env.ANALYTICS.writeDataPoint({ blobs: [pathname, method, cacheStatus, status], doubles: [latencyMs, 1], indexes: [userId] }); // 查询：按端点统计 P95 延迟 // SELECT blob1, quantile(0.95)(double1) AS p95_ms FROM perf GROUP BY blob1 ## 反模式 | ❌ 错误 | ✅ 正确 | |----------|-----------| | `await writeDataPoint()` | `writeDataPoint()` (即发即弃) | | `indexes: [method]` (低基数) | `blobs: [method]`, `indexes: [userId]` | | `blobs: [JSON.stringify(obj)]` | 在 blob 中存储 ID，完整对象存入 D1/KV | | 1000 万/分钟请求全部写入 | 每秒预聚合 | | 从 Worker 查询 | 从外部服务/API 查询 | ## 最佳实践 1. **预先设计 Schema** - 记录 blob/double/index 的分配 2. **始终包含计数指标** - `doubles: [latency, 1]` 用于 AVG 计算 3. **Blob 使用枚举** - 保持值的一致性，如 `Status.SUCCESS` 4. **处理采样** - 使用比率 (avg_latency = SUM(latency)/SUM(count)) 5. **尽早测试查询** - 在大量写入前验证 Schema ## Schema 模板 typescript /** * 数据集: my_metrics * * Blobs: * blob1: endpoint, blob2: method, blob3: status * * Doubles: * double1: latency_ms, double2: count (始终为 1) * * Indexes: * index1: customer_id (高基数) */