电商返利平台用户画像系统：Flink CDC实时数仓、标签体系与Lookalike扩量算法工程化

weixinscrm

371人浏览 · 2026-03-04 09:56:41

weixinscrm · 2026-03-04 09:56:41 发布

电商返利平台用户画像系统：Flink CDC实时数仓、标签体系与Lookalike扩量算法工程化

大家好，我是高佣返利省赚客APP研发者微赚！ 在存量竞争时代，精细化运营是电商返利平台生存的关键。传统的T+1离线画像已无法满足“秒级”营销决策的需求，用户刚浏览完母婴商品，下一秒就应收到奶粉优惠券。为此，省赚客APP研发团队构建了基于Flink CDC的实时数仓架构，打造了毫秒级更新的标签体系，并工程化落地了Lookalike（相似人群扩展）算法，实现了从数据接入到智能触达的全链路闭环。

一、基于Flink CDC的实时数据同步架构

传统ETL流程依赖日志解析，延迟高且易丢失。我们采用Flink CDC（Change Data Capture）技术，直接监听MySQL Binlog，将用户行为、订单状态及佣金变动实时同步至Kafka和Hudi数据湖，实现零延迟的数据摄入。

package cn.juwatech.flink.cdc.source;

import cn.juwatech.cn.model.UserBehaviorEvent;
import cn.juwatech.cn.schema.BehaviorSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.connector.mysql.cdc.MySqlSource;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;

public class RealTimeDataIngestion {

    public static DataStream<String> buildMySqlCdcStream(StreamExecutionEnvironment env) {
        MySqlSource<String> mysqlSource = MySqlSource.<String>builder()
            .hostname("rm-prod-mysql.internal")
            .port(3306)
            .databaseList("shengzhuanke_user_db")
            .tableList("shengzhuanke_user_db.user_behavior_log")
            .username("flink_cdc_user")
            .password("secure_password")
            .deserializer(new JsonDebeziumDeserializationSchema())
            .startupOptions(org.apache.flink.connector.mysql.cdc.source.enumerator.startup.StartupOptions.latest())
            .build();

        return env.fromSource(mysqlSource, WatermarkStrategy.noWatermarks(), "MySQL CDC Source")
            .map(record -> {
                // 解析CDC JSON数据，提取after字段作为最新状态
                return cn.juwatech.cn.util.CdcParser.extractAfterState(record);
            });
    }
}

通过上述代码，我们将数据库的变更流转化为Flink流处理引擎的输入源。无论是用户点击、加购还是下单，数据都能在秒级内进入计算管道，为实时标签更新奠定基础。

二、多维动态标签体系构建

基于实时流数据，我们构建了包含基础属性、行为偏好、消费能力及风险等级的四维标签体系。利用Flink的状态后端（State Backend），我们可以维护用户的时间窗口行为序列，动态计算标签值。

package cn.juwatech.flink.tag.computation;

import cn.juwatech.cn.model.UserTagProfile;
import cn.juwatech.cn.enums.TagCategory;
import cn.juwatech.cn.function.WindowAggregator;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class RealTimeTagUpdater extends KeyedProcessFunction<String, String, UserTagProfile> {

    private transient ValueState<UserTagProfile> tagState;

    @Override
    public void open(org.apache.flink.configuration.Configuration parameters) {
        tagState = getRuntimeContext().getState(
            new ValueStateDescriptor<>("user-tag-profile", UserTagProfile.class)
        );
    }

    @Override
    public void processElement(String eventJson, Context ctx, Collector<UserTagProfile> out) throws Exception {
        UserTagProfile currentProfile = cn.juwatech.cn.parser.EventParser.parse(eventJson);
        UserTagProfile existingProfile = tagState.value();

        if (existingProfile == null) {
            existingProfile = new UserTagProfile(ctx.getCurrentKey());
        }

        // 实时更新行为频次标签
        if ("ORDER_PAID".equals(currentProfile.getLastAction())) {
            existingProfile.incrementOrderCount();
            existingProfile.updateTotalGmv(currentProfile.getOrderAmount());
            // 动态打标：高价值用户
            if (existingProfile.getTotalGmv() > 5000.0) {
                existingProfile.addTag(TagCategory.HIGH_VALUE_USER);
            }
        }

        // 更新偏好标签：基于最近1小时浏览品类
        existingProfile.updateCategoryPreference(currentProfile.getCategoryId(), System.currentTimeMillis());

        tagState.update(existingProfile);
        out.collect(existingProfile);
    }
}

该逻辑确保了用户标签的实时性。一旦用户完成高额支付，其“高价值用户”标签立即生效，下游营销系统可瞬间触发VIP专属返利活动。

三、Lookalike相似人群扩量算法工程化

有了精准的种子用户画像，如何找到更多潜在高活用户？我们引入了Lookalike算法。通过将用户特征向量化为Embedding，计算种子用户与全量用户的余弦相似度，挖掘潜在目标群体。

package cn.juwatech.algorithm.lookalike;

import cn.juwatech.cn.model.VectorEmbedding;
import cn.juwatech.cn.service.VectorStoreClient;
import cn.juwatech.cn.dto.SimilarUserResult;
import java.util.List;
import java.util.stream.Collectors;

public class LookalikeEngine {

    private final VectorStoreClient vectorStore;

    public LookalikeEngine(VectorStoreClient client) {
        this.vectorStore = client;
    }

    /**
     * 基于种子用户群进行相似人群扩展
     * @param seedUserIds 种子用户ID列表（如：高复购母婴用户）
     * @param topK 返回相似用户数量
     */
    public List<SimilarUserResult> expandAudience(List<String> seedUserIds, int topK) {
        // 1. 获取种子用户的平均特征向量
        VectorEmbedding seedVector = aggregateSeedVectors(seedUserIds);

        // 2. 在向量数据库中检索Top-K近邻
        List<VectorEmbedding> neighbors = vectorStore.searchSimilarVectors(seedVector, topK * 2);

        // 3. 过滤已转化用户，计算最终得分
        return neighbors.stream()
            .filter(user -> !seedUserIds.contains(user.getUserId()))
            .filter(user -> !cn.juwatech.cn.service.ConversionService.hasConverted(user.getUserId()))
            .map(user -> {
                double similarity = calculateCosineSimilarity(seedVector, user);
                return new SimilarUserResult(user.getUserId(), similarity);
            })
            .sorted((a, b) -> Double.compare(b.getScore(), a.getScore()))
            .limit(topK)
            .collect(Collectors.toList());
    }

    private VectorEmbedding aggregateSeedVectors(List<String> userIds) {
        // 调用内部工具类计算向量均值
        return cn.juwatech.cn.math.VectorMath.meanVectors(
            userIds.stream().map(vectorStore::getVector).collect(Collectors.toList())
        );
    }

    private double calculateCosineSimilarity(VectorEmbedding v1, VectorEmbedding v2) {
        return cn.juwatech.cn.math.VectorMath.cosine(v1.getData(), v2.getData());
    }
}