如何基于时间戳获取产品最新状态并生成统计仪表盘？

核心需求 #

仅统计每个产品的最新状态，并按状态类型生成唯一产品数量统计，用于仪表盘展示。

示例日志数据 #

{
  "storeId": 3434,
  "products": [
    {
      "code": "my_code",
      "status": "http_issue"
    }
  ],
  "timestamp": "2025-02-13T10:48:01Z"
}
{
  "storeId": 3434,
  "products": [
    {
      "code": "my_code",
      "status": "cookie_issue"
    }
  ],
  "timestamp": "2025-02-14T10:48:01Z"
}
{
  "storeId": 3434,
  "products": [
    {
      "code": "my_code",
      "status": "product_issue"
    }
  ],
  "timestamp": "2025-02-15T10:48:01Z"
}

期望统计输出 #

total_unique_products_with_http_issues: 0
total_unique_products_with_cookie_issues: 0
total_unique_products_with_product_issues: 1

最佳实现方案 #

通用逻辑 #

1. 按产品分组取最新记录：以产品code为维度，筛选出每个产品对应的最新timestamp记录，确保只保留当前有效状态。
2. 按状态统计唯一产品数：基于筛选后的最新记录，按status分组统计不同状态下的唯一产品数量。

1. SQL数据库实现（适用于日志存储在数据库的场景） #

假设日志存储在logs表中，products字段为数组类型，可通过以下SQL实现：

-- 第一步：获取每个产品的最新时间戳
WITH latest_product_records AS (
    SELECT 
        p.code,
        MAX(l.timestamp) AS latest_ts
    FROM logs l
    -- 展开products数组（不同数据库语法有差异：PostgreSQL用UNNEST，MySQL用JSON_TABLE）
    CROSS JOIN UNNEST(l.products) p
    GROUP BY p.code
)
-- 第二步：关联原表获取最新状态并统计
SELECT
    -- 格式化指标名称
    CONCAT('total_unique_products_with_', status, '_issues') AS metric_name,
    COUNT(DISTINCT p.code) AS metric_value
FROM logs l
CROSS JOIN UNNEST(l.products) p
JOIN latest_product_records lpr 
    ON p.code = lpr.code AND l.timestamp = lpr.latest_ts
GROUP BY status
-- 补充未出现的状态，确保输出完整
UNION ALL
SELECT 'total_unique_products_with_http_issues', 0 
WHERE NOT EXISTS (SELECT 1 FROM logs l CROSS JOIN UNNEST(l.products) p WHERE status = 'http_issue')
UNION ALL
SELECT 'total_unique_products_with_cookie_issues', 0 
WHERE NOT EXISTS (SELECT 1 FROM logs l CROSS JOIN UNNEST(l.products) p WHERE status = 'cookie_issue')
UNION ALL
SELECT 'total_unique_products_with_product_issues', 0 
WHERE NOT EXISTS (SELECT 1 FROM logs l CROSS JOIN UNNEST(l.products) p WHERE status = 'product_issue')
ORDER BY metric_name;

2. Python脚本实现（适用于本地日志或轻量数据处理） #

直接读取日志文件，处理后输出统计结果：

import json
from collections import defaultdict

# 读取日志（每行一个JSON对象）
product_logs = []
with open("product_status_logs.json", "r") as f:
    for line in f:
        if line.strip():
            product_logs.append(json.loads(line))

# 存储每个产品的最新状态
latest_status = {}

for log_entry in product_logs:
    current_ts = log_entry["timestamp"]
    for product in log_entry["products"]:
        prod_code = product["code"]
        # 更新最新状态：产品不存在或当前时间戳更新
        if prod_code not in latest_status or current_ts > latest_status[prod_code]["timestamp"]:
            latest_status[prod_code] = {
                "status": product["status"],
                "timestamp": current_ts
            }

# 统计各状态的产品数量
status_counter = defaultdict(int)
for prod_data in latest_status.values():
    status_counter[prod_data["status"]] += 1

# 按期望格式输出
metric_mapping = {
    "http_issue": "total_unique_products_with_http_issues",
    "cookie_issue": "total_unique_products_with_cookie_issues",
    "product_issue": "total_unique_products_with_product_issues"
}

for status_key, metric_name in metric_mapping.items():
    print(f"{metric_name}: {status_counter.get(status_key, 0)}")

3. 仪表盘集成建议 #

• 如果使用Grafana：将SQL查询结果作为数据源，配置单值面板或表格面板展示统计指标；
• 如果需要监控告警：可将Python脚本的输出转换为Prometheus自定义指标，再通过Grafana配置可视化和告警规则；
• 状态枚举：提前定义所有可能的状态值，避免统计遗漏。

内容的提问来源于stack exchange，提问作者Xus