---

背景与痛点

全球气象数据采集面临三大核心挑战：

1. 数据延迟问题：观测站分布在不同时区，网络条件差异大，部分偏远地区数据传输延迟可能超过1小时
2. 高并发压力：全球数万个观测站每分钟上报数据，高峰期需处理每秒上千条写入请求
3. 存储成本激增：单站每天产生1440条记录，10年数据量可达525万条/站，传统关系型数据库难以支撑

技术选型

消息队列对比

• Kafka：高吞吐(每秒百万级消息)、持久化存储、支持分区和消费者组，适合日志类数据
• RabbitMQ：低延迟(毫秒级)、复杂路由机制，但吞吐量约5万/秒
• RocketMQ：阿里系方案，事务消息支持更好

最终选择Kafka 3.0+版本，因其：

1. 天然支持地理分布式部署
2. 消息保留策略可配置(我们设置7天)
3. 与Flink等流处理框架无缝集成

数据库选型

| 方案 | 写入速度 | 压缩比 | 查询性能 | 成本 | |---------------|------------|--------|----------|--------| | InfluxDB | 50万点/秒 | 10:1 | ★★★★ | 中 | | TimescaleDB | 30万点/秒 | 7:1 | ★★★★★ | 较高 | | ClickHouse | 200万点/秒 | 20:1 | ★★★ | 低 |

选择TimescaleDB(PostgreSQL扩展)因其：

1. 完整SQL支持，便于与其他系统集成
2. 自动分块(Chunk)管理时序数据
3. 支持连续聚合(Continuous Aggregate)

架构设计

核心组件：

1. 采集层：驻留在各观测站的Agent程序，通过MQTT协议上报数据
2. 接入层：Kafka集群，按大洲划分Topic分区(asia.temp, europe.temp等)
3. 处理层：Flink实时清洗数据(过滤异常值、单位转换)
4. 存储层：TimescaleDB分片集群，按年份分表
5. 服务层：Spring Cloud微服务提供API

核心实现

数据采集模块(Python示例)

# 观测站Agent
import paho.mqtt.client as mqtt
from datetime import datetime

def on_connect(client, userdata, flags, rc):
    print("Connected with result code "+str(rc))

client = mqtt.Client(client_id="station_1234")
client.on_connect = on_connect
client.connect("kafka-broker.example.com", 1883, 60)

while True:
    temp = read_sensor()  # 获取传感器数据
    payload = {
        "station_id": "1234",
        "timestamp": datetime.utcnow().isoformat(),
        "value": temp,
        "geo": "39.9042,116.4074"
    }
    client.publish("asia.temp", json.dumps(payload))
    time.sleep(60)  # 每分钟发送

流处理模块(Java Flink示例)

// 温度数据处理流水线
DataStream<StationRecord> records = env
    .addSource(new KafkaSource<>("asia.temp"))
    .map(json -> JSON.parseObject(json, StationRecord.class))
    .filter(record -> {
        // 过滤异常值
        return record.getValue() > -60 && record.getValue() < 60; 
    })
    .keyBy(StationRecord::getStationId)
    .process(new TemperatureAlertProcess());

// 写入TimescaleDB
records.addSink(JdbcSink.sink(
    "INSERT INTO temperatures VALUES (?, ?, ?, ?)",
    (stmt, record) -> {
        stmt.setString(1, record.getStationId());
        stmt.setTimestamp(2, record.getTimestamp());
        stmt.setDouble(3, record.getValue());
        stmt.setString(4, record.getGeo());
    },
    new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
        .withUrl("jdbc:postgresql://tsdb:5432/weather")
        .withDriverName("org.postgresql.Driver")
        .withUsername("ingest")
        .withPassword("password")
        .build()
));

性能优化

批处理策略

1. Kafka生产者：设置linger.ms=500和batch.size=16384，减少小包传输
2. TimescaleDB写入：使用COPY命令批量导入，相比INSERT提升30倍吞吐

压缩配置

# Kafka配置
compression.type: zstd  # 比gzip节省15%空间
log.segment.bytes: 1073741824  # 1GB分段

# TimescaleDB配置
timescaledb.compress:
  segment_by: station_id
  chunk_time_interval: 7d
  compression_algorithm: zstd

缓存设计

• Redis缓存热点数据：最近12小时数据+频繁查询站点
• 本地Caffeine缓存：各微服务实例缓存常用查询

生产环境考量

监控体系

1. 指标采集：Prometheus收集各组件Metrics
2. Kafka：消息堆积量、分区延迟
3. Flink：Checkpoint时长、反压指标
4. TimescaleDB：查询耗时、压缩率
5. 告警规则：当某分区延迟>5分钟触发SMS通知

容错机制

• Kafka：设置replication.factor=3和min.insync.replicas=2
• Flink：开启Checkpoint，设置EXACTLY_ONCE语义
• 数据补传：各观测站本地存储最近7天原始数据

避坑指南

1. 时区问题：
2. 所有时间戳必须使用UTC
3. 在查询层做时区转换
4. 磁盘写满：
5. 为TimescaleDB设置磁盘水位线监控
6. 配置自动删除旧分片策略
7. Schema变更：
8. 新增字段时使用ALTER TABLE...ADD COLUMN IF NOT EXISTS
9. 避免频繁修改字段类型

扩展思考

1. 如何设计降级方案应对跨国网络中断？
2. 当需要回溯修正历史数据时，系统架构需要哪些调整？
3. 对于气象数据这类强时空关联的数据，有哪些更高效的查询优化手段？

这个系统已稳定运行18个月，日均处理2.3亿条温度记录。最重要的经验是：在数据接入层保持简单，把复杂性下推到处理层。这种分层架构让我们能够独立扩展每个组件，比如去年双十一期间临时增加了5个Flink TaskManager实例应对流量高峰。