维度表离线和实时都要用，维护两份又怕不一致——维度服务怎么设计？

上一篇讲了口径一致性，其中有一个关键细节——维度表的一致性。在Lambda架构中，离线链路的维度表在Hive里，实时链路的维度表在Redis或MySQL里，两边怎么保持同步？如果渠道表更新了一个新渠道，离线能查到但实时查不到，JOIN出来就是NULL。这种问题我在某短剧出海公司遇到过好几次，最终设计了一套统一的维度服务方案，今天分享出来。

维度服务整体架构

我在某短剧出海公司设计的维度服务架构如下：

维度数据流向（一个源头、多端同步）：
 
                    ┌─→ Hive/MaxCompute（离线DWD JOIN用）
                    │   每天全量同步一次
                    │
MySQL（唯一源头）──→ Canal ──→ Kafka ─┤
                    │   ├─→ Redis（实时Flink JOIN用）
                    │   │   实时增量同步
                    │   │
                    │   └─→ Hologres（OLAP查询用）
                    │       实时增量同步
                    │
                    └─→ DataWorks全量同步（兜底）
                        每天凌晨全量覆盖，确保一致性

核心思路是：MySQL的变更通过Canal采集到Kafka，Kafka消费端分别写入Redis和Hologres（实时同步）；同时每天凌晨通过DataWorks做一次全量同步到Hive和Hologres（离线兜底）。这样无论维度表怎么更新，各端在几秒到几分钟内都能看到最新数据。

方案一：预加载到Flink State（适合小维度表）

如果维度表很小（几千到几万行，比如渠道表、国家表、币种表），最简单的方案是启动时全量加载到Flink的State中，然后定期刷新。

// Flink预加载维度表到State
public class DimChannelFunction extends RichMapFunction<Event, EnrichedEvent> {
    private Map<String, ChannelInfo> channelCache;
 
    @Override
    public void open(Configuration params) {
        // 启动时从MySQL全量加载渠道维度表
        channelCache = loadFromMySQL("SELECT * FROM dim_channel");
        // 注册定时器，每5分钟刷新一次
        getRuntimeContext().getScheduledExecutor()
            .scheduleAtFixedRate(this::refreshCache, 5, 5, TimeUnit.MINUTES);
    }
 
    @Override
    public EnrichedEvent map(Event event) {
        ChannelInfo channel = channelCache.get(event.getChannelId());
        return new EnrichedEvent(event, channel);  // 维度退化
    }
 
    private void refreshCache() {
        channelCache = loadFromMySQL("SELECT * FROM dim_channel");
    }
}

适用场景：维度表<10万行、更新频率低（每天几次）、不需要精确的版本控制。我们的渠道表只有50行，国家表200多行，用这种方案最省事。

不适用：维度表太大（百万行+）会占用大量Task内存；更新非常频繁（每秒级别）会导致MySQL压力大。

方案二：旁路查Redis缓存（适合中等维度表）

如果维度表有几十万到几百万行，预加载到State不现实，可以用Flink的异步IO查Redis。维度数据由同步任务实时写入Redis，Flink在处理每条事件时异步查询Redis获取维度信息。

// Flink异步IO查Redis维度表
public class AsyncDimLookup extends RichAsyncFunction<Event, EnrichedEvent> {
    private transient RedisAsyncCommands<String, String> redis;
 
    @Override
    public void open(Configuration params) {
        RedisClient client = RedisClient.create("redis://redis-host:6379");
        redis = client.connect().async();
    }
 
    @Override
    public void asyncInvoke(Event event, ResultFuture<EnrichedEvent> future) {
        // 异步查Redis：key = dim:user:{user_id}
        String key = "dim:user:" + event.getUserId();
        redis.hgetall(key).thenAccept(fields -> {
            UserInfo user = parseUserInfo(fields);
            future.complete(Collections.singleton(
                new EnrichedEvent(event, user)));
        }).exceptionally(e -> {
            // Redis查询失败时用默认值，不要阻塞流
            future.complete(Collections.singleton(
                new EnrichedEvent(event, UserInfo.UNKNOWN)));
            return null;
        });
    }
}