淘客返利app的数据中台架构：数据资产化与服务化

虫小宝

324人浏览 · 2025-09-26 07:30:00

虫小宝 · 2025-09-26 07:30:00 发布

淘客返利app的数据中台架构：数据资产化与服务化

大家好，我是阿可，微赚淘客系统及省赚客APP创始人，是个冬天不穿秋裤，天冷也要风度的程序猿！

淘客返利APP日均产生千万级订单数据、亿级用户行为日志，分散在业务数据库、缓存、日志文件中，形成“数据孤岛”。数据中台通过资产化梳理与服务化封装，将碎片化数据转化为可复用的数据服务，支撑运营分析、个性化推荐等核心场景。本文结合省赚客APP实践，拆解数据中台的技术实现。
淘客返利app

一、架构设计：三层架构支撑数据全生命周期

数据中台采用“数据采集-资产化处理-服务化输出”三层架构，核心技术栈如下：

采集层：Flink CDC + Logstash，实现业务数据实时同步与日志采集
处理层：Hive + Spark，负责数据清洗、建模与资产化沉淀
服务层：Spring Cloud + ClickHouse，提供低延迟数据查询服务

核心目标是建立统一数据模型，将“订单、用户、商品、返利”四类核心数据转化为可复用的数据资产。

二、数据资产化：从原始数据到资产的落地实现

2.1 实时数据采集模块

基于Flink CDC实现MySQL业务数据实时同步，避免侵入业务系统。

package cn.juwatech.data.collector;

import cn.juwatech.data.config.FlinkEnvConfig;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class MysqlDataCollector {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = FlinkEnvConfig.getEnv();

        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("mysql-master")
                .port(3306)
                .databaseList("taoke_order", "taoke_user") // 同步订单、用户库
                .tableList("taoke_order.t_order", "taoke_user.t_user") // 指定表
                .username("cdc_user")
                .password("cdc_pass123")
                .deserializer(new JsonDebeziumDeserializationSchema()) // 输出JSON格式
                .build();

        env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql-cdc-source")
                .addSink(new KafkaSink("data-collect-topic")) // 写入Kafka
                .name("mysql-data-sink");

        env.execute("taoke-mysql-cdc-collect");
    }
}

2.2 数据建模与资产注册

采用维度建模方法构建数据仓库，通过资产注册中心统一管理数据资产元信息。

package cn.juwatech.data.asset;

import cn.juwatech.data.model.DataAsset;
import cn.juwatech.data.model.AssetType;
import cn.juwatech.data.mapper.AssetMetaMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class AssetRegistryService {

    @Autowired
    private AssetMetaMapper assetMetaMapper;

    // 注册订单主题数据资产
    public void registerOrderAsset() {
        DataAsset asset = new DataAsset();
        asset.setAssetName("订单核心资产");
        asset.setAssetCode("TAOKE_ORDER_ASSET_001");
        asset.setAssetType(AssetType.FACT_TABLE);
        asset.setDataSource("hive.taoke_ods.t_order_dwd");
        asset.setDescription("包含订单基本信息、返利金额、商品关联等核心字段");
        asset.setOwner("数据团队");
        // 注册字段元信息
        asset.setFields("[{\"fieldName\":\"order_id\",\"type\":\"string\",\"comment\":\"订单ID\"},{\"fieldName\":\"rebate_amount\",\"type\":\"bigint\",\"comment\":\"返利金额\"}]");
        
        assetMetaMapper.insert(asset);
    }
}

2.3 数据质量监控实现

通过自定义规则校验数据完整性，确保资产可靠性。

package cn.juwatech.data.quality;

import cn.juwatech.data.mapper.OrderDwdMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@Component
public class DataQualityChecker {

    @Autowired
    private OrderDwdMapper orderDwdMapper;
    @Autowired
    private QualityAlarmService alarmService;

    // 每小时校验订单数据完整性
    @Scheduled(cron = "0 0 * * * ?")
    public void checkOrderDataComplete() {
        // 1. 校验订单ID非空
        long nullOrderIdCount = orderDwdMapper.countNullOrderId();
        if (nullOrderIdCount > 0) {
            alarmService.sendAlarm("订单资产校验失败", "订单ID为空数量：" + nullOrderIdCount);
        }

        // 2. 校验返利金额合理性
        long abnormalRebateCount = orderDwdMapper.countAbnormalRebate();
        if (abnormalRebateCount > 0) {
            alarmService.sendAlarm("订单资产校验失败", "返利金额异常数量：" + abnormalRebateCount);
        }
    }
}

三、数据服务化：资产到应用的高效输出

3.1 数据服务接口开发

基于Spring Boot封装数据资产为REST接口，供业务系统调用。

package cn.juwatech.data.service.controller;

import cn.juwatech.data.service.dto.UserRebateDTO;
import cn.juwatech.data.service.service.UserRebateService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
@Api(tags = "用户返利数据服务")
public class UserRebateDataController {

    @Autowired
    private UserRebateService userRebateService;

    @GetMapping("/api/data/user/rebate")
    @ApiOperation("查询用户累计返利")
    public UserRebateDTO getUserTotalRebate(@RequestParam String userId) {
        // 从ClickHouse查询预处理后的资产数据
        return userRebateService.getTotalRebateByUserId(userId);
    }

    @GetMapping("/api/data/user/rebate/trend")
    @ApiOperation("查询用户返利趋势")
    public List<RebateTrendDTO> getUserRebateTrend(
            @RequestParam String userId,
            @RequestParam String startDate,
            @RequestParam String endDate) {
        return userRebateService.getRebateTrend(userId, startDate, endDate);
    }
}

3.2 服务熔断与缓存优化

为保障服务稳定性，添加熔断与缓存机制。

package cn.juwatech.data.service.config;

import cn.juwatech.data.service.service.UserRebateService;
import com.alibaba.csp.sentinel.annotation.aspectj.SentinelResourceAspect;
import com.alibaba.csp.sentinel.slots.block.RuleConstant;
import com.alibaba.csp.sentinel.slots.block.flow.FlowRule;
import com.alibaba.csp.sentinel.slots.block.flow.FlowRuleManager;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;

@Configuration
public class DataServiceConfig {

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Bean
    public SentinelResourceAspect sentinelResourceAspect() {
        return new SentinelResourceAspect();
    }

    // 初始化流控规则
    @PostConstruct
    public void initFlowRules() {
        List<FlowRule> rules = new ArrayList<>();
        FlowRule rule = new FlowRule();
        rule.setResource("getUserTotalRebate");
        rule.setGrade(RuleConstant.FLOW_GRADE_QPS);
        rule.setCount(100); // 限制QPS为100
        rules.add(rule);
        FlowRuleManager.loadRules(rules);
    }

    // 缓存切面，缓存用户返利数据
    @Bean
    public UserRebateCacheAspect userRebateCacheAspect() {
        return new UserRebateCacheAspect(redisTemplate);
    }
}