技术科普 | 基于 Flink + Doris 体验实时数仓建设

为了更好地理解实时数仓的建设流程，在详细介绍其构建方法之前，先为大家科普下实时计算与实时数仓之间的联系与区别。

• 实时计算

普通的实时计算会优先考虑时效性，采集数据源后，经过实时计算直接得到计算结果。这样虽保证了时效性，但可能会存在大量的重复计算，导致更多的计算资源开销。

• 实时数仓

实时数仓基于一定的数据仓库理念，对数据处理流程进行规划、分层，目的是提高数据和计算的复用性。在一定程度上减少了数据的重复计算，计算流程上会存在更多依赖关系，因此数据的时效性会略差一点。

• ODS层
  

原始数据层：也称贴源层、操作数据层，数据是最细粒度的，没做任何变动，如行为日志、系统日志和业务数据。

• DWD层

明细数据层：保持和ODS层一样的数据颗粒度，主要是对ODS层数据做清洗和规范化操作，比如数据类型转换、数据对齐、数据过滤等。

• DIM层(Hbase)

维度数据层：建立标准数据分析维度，可以降低数据计算口径和算法不统一风险，如时间维度、地区维度、产品维度、机构维度等。

• DWM层

中间数据层：在DWD层的数据基础上，对数据做一些轻度聚合或关联操作，生成一些列的中间结果表，提升公共指标的复用性，减少重复加工处理，也可以和维度进行关联，形成宽表，其数据仍然是明细数据。

• DWS层

汇总数据层：基于DWD或者DWM上的基础数据，根据主题划分将多个事实数据按维度进行聚合汇总，形成主题宽表。用于后续的业务查询、OLAP分析、数据分发等。

• ADS层
  

应用数据层：提供给数据产品和数据分析使用，把OLAP分析引擎中的数据根据需要进行聚合筛选，并提供API接口服务。

从上图DWS层可以看到，实时数据分析引擎存储可以是多种组合，可以选择ClickHouse或者Apache Doris，甚至可以是多种组件的组合，由此看出实时数仓构建方案的多样灵活，选择哪种实现方案，主要还要根据各自应用场景而定，没有哪一种OLAP引擎是万能的，比较常见的组合如下：

• Kafka + Flink + ClickHouse 简称KFC

• Kafka + Flink + Doris 简称KFD

• Kafka + Flink + 其他实时OLAP引擎

Doris和ClickHouse两种OLAP引擎都具备一定的优势，分别如下：

通过以上分析比较，本案例为基于简单的业务场景体验实时数仓的建设流程，因此选择KFD方案实现更加合适。且Doris支持事物和幂等写入，与Flink结合能更好地实现数据精准一次性（Exactly-Once）处理。

前文的案例简介中已明确描述，以应用访问的行为日志进行流量分析，从简单的对应用访问PV、UV功能入手，一步步探索实时数仓构建的流程。

首先，准备数据源，为方便起见，对后文用到的数据结构进行简单定义。

• 行为数据定义
  

行为日志是直接写到Kafka，作为数仓的ODS层，创建Kafka主题(topic)并按照ODS层的命名规则命名为ods_base_log，表结构如下所示：

数据结构如下所示：

• 维度数据定义
  

行为日志中只包含了应用的id，为了体现出数仓的维度聚合，用Flink CDC将MySQL的配置数据（应用列表）进行同步，将数据写入Kafka的ODS层，命名为ods_base_db。然后通过下一步处理，读取ODS层ods_base_db主题中的数据，并筛选出app_list维度数据，写到DIM层，存储于HBASE中，命名为dim_app_list，表结构如下所示：

数据结构如下所示：

对于行为数据收集，用SpringBoot实现一个Kafka生产者的日志写入服务，对应用服务访问时产生的行为日志，通过调用生产者服务接口写入到ods_base_log主题，如下所示：

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
@RestController
@Slf4j
public class LoggerController {
  
    @Autowired
    KafkaTemplate kafkaTemplate;
  
    @RequestMapping("/applog")
    public String logger(@RequestParam("param")String param){
        // 日志数据进入ODS往ods_base_log主题写入
        // 业务数据进入ODS往ods_base_db 主题写入
        kafkaTemplate.send("ods_base_log", param);
        return param;
    }
}

由于系统业务数据及维度数据都存储在业务数据库中，为了能实时捕获表的数据变动，则通过Flink CDC从MySQL（或MongoDB，由实际业务系统应用情况而定）中读取全库数据或部分表，并写入到Kafka的ods_base_db主题，简单的实现方式如下所示：

import com.kingdee.common.DmpConfig;
import com.kingdee.function.MyDeserialization;
import com.kingdee.utils.MyKafkaUtil;
import com.ververica.cdc.connectors.mysql.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.DebeziumSourceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class FlinkCDCWithMyDeserialization {
  
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DebeziumSourceFunction<String> sourceFunction = MySqlSource.<String>builder()
                .hostname(DmpConfig.MYSQL_HOST)
                .username(DmpConfig.MYSQL_USERNAME)
                .password(DmpConfig.MYSQL_PSW)
                .databaseList("busi_db")
                // 如果不添加该参数，则消费指定数据库中所有表的数据
                // 如果指定，则消费指定表的数据，表名必须带上库名
                //.tableList("busi_db.*")
                .deserializer(new MyDeserialization())
                //.debeziumProperties()
                .startupOptions(StartupOptions.latest())
                .build();
    
        //addSource
        DataStreamSource<String> streamSource = env.addSource(sourceFunction);
    
        streamSource.print("streamSource>>>>>");
        // 业务数据写入 ods_base_db 主题
        // 日志数据写入 ods_base_log 主题
        String sinkTopic = "ods_base_db";
        streamSource.addSink(MyKafkaUtil.getKafkaProducer(sinkTopic));
    
        env.execute("FlinkCDCWithMyDeserialization");
    }
}

String topic = "ods_base_log";
String groupId = "base_log_app";
DataStreamSource<String> kafkaDs = env.addSource(MyKafkaUtil.getKafkaConsumer(topic, groupId));
OutputTag<String> outputTag = new OutputTag<String>("ErrDataTag"){};
OutputTag<String> otherOutputTag = new OutputTag<String>("other"){};
SingleOutputStreamOperator<JSONObject> jsonObjDs = kafkaDs.process(new ProcessFunction<String, JSONObject>() {
@Override
public void processElement(String s, Context context, Collector<JSONObject> collector) throws Exception {
  try {
    JSONObject json = JSON.parseObject(s);
    if("prd".equals(json.getString("dataType"))){
      collector.collect(json);
    }else{
      context.output(otherOutputTag, s);
    }
  } catch (Exception e) {
    //发生异常，将数据写入测输出流
    context.output(outputTag, s);
}
}
});
jsonObjDs.getSideOutput(outputTag).print("Parse Err Data>>>");
jsonObjDs.getSideOutput(otherOutputTag).print("Other Data>>>");
jsonObjDs.map(r -> r.toJSONString()).addSink(MyKafkaUtil.getKafkaProducer("dwd_pv_log"));

2、读取Kafka数据源，处理ods_base_db数据，并将数据转换为JSON格式，同时过滤掉删除操作的记录，保留新增、修改的记录，此步包括了业务数据和维度数据的处理，将不同的数据类型按照配置进行分发。

String topic = "ods_base_db";
String groupId = "base_db_app";
DataStreamSource<String> kafkaDs = env.addSource(MyKafkaUtil.getKafkaConsumer(topic, groupId));    
SingleOutputStreamOperator<JSONObject> jsonObjDs = kafkaDs.map(JSON::parseObject)
  .filter(new FilterFunction<JSONObject>() {
    @Override
    public boolean filter(JSONObject value) throws Exception {
      //取出数据的操作类型
      String type = value.getString("type");
      return !"delete".equals(type);
    }
  });

3、由于主流中包含了所有的业务数据与维度数据，要明确数据处理后的具体流向，因此，可以事先定义一个配置表，用于管理源头表的元数据，如：表类型，数据流向目标等信息，最后通过Flink CDC读取数据配置表数据，并将其转换为广播流，数据保存在状态变量中，便于后续流处理读取配置数据。实现如下：

DebeziumSourceFunction<String> sourceFunction = MySqlSource.<String>builder()
        .hostname(DmpConfig.MYSQL_HOST)
        .port(DmpConfig.MYSQL_PORT)
        .username(DmpConfig.MYSQL_USERNAME)
        .password(DmpConfig.MYSQL_PSW)
        .databaseList("busi_db")
        .tableList("meta_config")
        .startupOptions(StartupOptions.initial())
        .deserializer(new MyDeserialization())
        .build();
DataStreamSource<String> configDs = env.addSource(sourceFunction);
MapStateDescriptor<String, DataConfigProcess> mapStateDescriptor = new MapStateDescriptor<>(
        "map-state",
        String.class,
        DataConfigProcess.class);
BroadcastStream<String> broadcastStream = configDs.broadcast(mapStateDescriptor);

从数据分析得知，统计PV、UV用到的源数据都是ods_base_log，PV统计的是时间范围内访问的次数，对每次访问进行累计，而UV统计的是独立访问用户数，在本案例中指时间范围内访问的用户数量。

由此可得，UV可以通过PV的记录去重后获取，直接在DWD层的dwd_pv_log基础上去重即可，避免了UV统计时读取ods_base_log源数据进行重新统计。实现方法如下：

1、通过Flink读取dwd_pv_log数据，转换成JSON格式，并提取watermark，如下所示：

String sourceTopic = "dwd_pv_log";
String groupId = "unique_visit_app";
String sinkTopic = "dwm_uv_log";
DataStreamSource<String> kafkaDs = env.addSource(MyKafkaUtil.getKafkaConsumer(sourceTopic, groupId));
SingleOutputStreamOperator<JSONObject> jsonObjDs = kafkaDs.map(JSON::parseObject).assignTimestampsAndWatermarks(
WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3L))
      .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
        @Override
        public long extractTimestamp(JSONObject value, long timestamp) {
          return value.getLong("logTs");
        }
      })
  );

2、按应用维度（appId）与用户id（userId）进行keyBy获得KeyedStream，实现分组功能，如下所示：

String sourceTopic = "dwd_pv_log";