如何将Landing DB数据迁移至Staging PostgreSQL并筛选最优徽章数据

方案一：基于Kafka流处理直接过滤（推荐） #

既然数据已经经过Kafka链路，完全可以在Kafka层直接完成过滤计算，无需等数据落地到Landing DB再处理，从根源减少IO和计算开销：

• 用Kafka Streams或Apache Flink消费MQTT转Kafka的原始数据：
  • 按网关ID+时间间隔窗口做分组（比如5分钟窗口）
  • 对每个窗口内的同网关数据，用聚合函数筛选出距离最小的徽章MAC地址
  • 直接将计算结果写入Staging DB，同时保留原始数据到Landing DB
• 核心优势：
  • 流式计算为实时/准实时处理，避免Cronjob批量调度带来的延迟和资源突增问题
  • 彻底规避数据库层复杂计算的性能瓶颈，计算逻辑与存储解耦，便于后续调整时间间隔或过滤规则

方案二：优化PostgreSQL批量处理（兼容现有架构） #

若不想改动现有Kafka链路，可针对原有存储过程+Cronjob做针对性优化：

• 替换存储过程为批量SQL语句，避免游标或逐行处理的低效逻辑：

  INSERT INTO staging_db.target_table (gateway_id, badge_mac, window_time, min_distance)
  SELECT 
    gateway_id, 
    badge_mac, 
    date_trunc('5 minutes', timestamp) AS window_time, 
    MIN(distance)
  FROM landing_db.source_table
  WHERE timestamp >= (SELECT COALESCE(MAX(window_time), '1970-01-01') FROM staging_db.target_table)
  GROUP BY gateway_id, badge_mac, date_trunc('5 minutes', timestamp);
  

• 调整Cronjob调度策略：
  • 缩短调度间隔（比如从1小时改为10分钟），减少单次处理的数据量
  • 给landing_db.source_table的timestamp和gateway_id字段建立联合索引，加速分组查询
• 核心优势：改动最小，兼容现有架构；批量SQL执行效率远高于传统存储过程

方案三：基于CDC的增量处理 #

利用PostgreSQL的CDC（变更数据捕获）功能，监听Landing DB的新增数据，实时触发过滤逻辑：

• 开启PostgreSQL的wal_level = logical配置，用Debezium捕获Landing DB的新增数据事件
• 将捕获到的增量数据发送到Kafka，再通过流处理组件（同方案一）做窗口聚合，最终写入Staging DB
• 核心优势：既保留数据落地到Landing DB的流程，又能实现准实时过滤迁移，避免批量处理的性能波动

方案选型建议 #

• 追求极致性能和实时性：优先选方案一，直接在数据流中间完成计算，减少不必要的数据落地开销
• 现有架构改动成本高：选方案二，快速解决性能问题
• 需保留Landing DB落地流程且要求准实时处理：选方案三

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