基于flink sql构建实时数据仓库
发布网友
发布时间：2022-09-04 18:11
共1个回答
热心网友
时间：2023-10-09 14:31
离线数仓的架构图：
实时数仓架构图:
目前是将实时维度表和DM层数据存于hbase当中，实时公共层都存于kafka当中，并且以写滚动日志的方式写入HDFS（主要是用于校验数据）。其实在这里可以做的工作还有很多，kafka集群，flink集群，hbase集群相互独立，这对整个实时数据仓库的稳定性带来一定的挑战。
一个数据仓库想要成体系，成资产，离不开数据域的划分。所以参考着离线的数据仓库，想着在实时数仓做出这方面的探索，理论上来讲，离线可以实现的，实时也是可以实现的。 并且目前已经取得了成效，目前划分的数据域跟离线大致相同，有流量域，交易域，营销域等等。当然这里面涉及到维表，多事务事实表，累计快照表，周期性快照表的设计，开发，到落地这里就不详述了。
维度表也是整个实时数据仓库不可或缺的部分。从目前整个实时数仓的建设来看，维度表有着数据量大，但是变更少的特点，我们试想过构建全平台的实时商品维度表或者是实时会员维度表，但是这类维度表太过于复杂，所以针对这类维度表下面介绍。还有另外一种就是较为简单的维度表，这类维度可能对应着业务系统单个mysql表，或者只需要几个表进行简单ETL就可以产出的表，这类维表是可以做成实时的。以下有几个实施的关键点：
如下是离线数据同步架构图：
实时数据的接入其实在底层架构是一样的，就是从kafka那边开始不一样，实时用flink的UDTF进行解析，而离线是定时（目前是小时级）用camus拉到HDFS，然后定时load HDFS的数据到hive表里面去，这样来实现离线数据的接入。实时数据的接入是用flink解析kafka的数据，然后在次写入kafka当中去。
由于目前离线数据已经稳定运行了很久，所以实时接入数据的校验可以对比离线数据，但是离线数据是小时级的hive数据，实时数据存于kafka当中，直接比较不了，所以做了相关处理，将kafka的数据使用flink写HDFS滚动日志的形式写入HDFS，然后建立hive表小时级定时去load HDFS中的文件，以此来获取实时数据。
完成以上两点，剩余还需要考虑一点，都是小时级的任务，这个时间卡点使用什么字段呢?首先要确定一点就是离线和实时任务卡点的时间字段必须是一致的，不然肯定会出问题。目前离线使用camus从kafka将数据拉到HDFS上，小时级任务，使用nginx_ts这个时间字段来卡点，这个字段是上报到nginx服务器上记录的时间点。而实时的数据接入是使用flink消费kafka的数据，在以滚动日志的形式写入HDFS的，然后在建立hive表load HDFS文件获取数据，虽然这个hive也是天/小时二级分区，但是离线的表是根据nginx_ts来卡点分区，但是实时的hive表是根据任务启动去load文件的时间点去区分的分区，这是有区别的，直接筛选分区和离线的数据进行对比，会存在部分差异，应当的做法是筛选范围分区，然后在筛选nginx_ts的区间，这样在跟离线做对比才是合理的。
目前实时数据接入层的主要时延是在UDTF函数解析上，实时的UDTF函数是根据上报的日志格式进行开发的，可以完成日志的解析功能。
解析流程图如下：
解析速率图如下：
该图还不是在峰值数据量的时候截的，目前以800记录/second为准，大概一个记录的解析速率为1.25ms。
目前该任务的flink资源配置核心数为1，假设解析速率为1.25ms一条记录，那么峰值只能处理800条/second，如果数据接入速率超过该值就需要增加核心数，保证解析速率。
介绍一下目前离线维度表的情况，就拿商品维度表来说，全线记录数将近一个亿，计算逻辑来自40-50个ods层的数据表，计算逻辑相当复杂，如果实时维度表也参考离线维度表来完成的话，那么开发成本和维护成本非常大，对于技术来讲也是很大的一个挑战，并且目前也没有需求要求维度属性百分百准确。所以目前（伪实时维度表）准备在当天24点产出，当天的维度表给第二天实时公共层使用，即T-1的模式。伪实时维度表的计算逻辑参考离线维度表，但是为了保障在24点之前产出，需要简化一下离线计算逻辑，并且去除一些不常用的字段，保障伪实时维度表可以较快产出。
实时维度表的计算流程图：
目前使用flink作为公司主流的实时计算引擎，使用内存作为状态后端，并且固定30s的间隔做checkpoint，使用HDFS作为checkpoint的存储组件。并且checkpoint也是作为任务restart以后恢复状态的重要依据。熟悉flink的人应该晓得，使用内存作为状态后端，这个内存是JVM的堆内存，毕竟是有限的东西，使用不得当，OOM是常有的事情，下面就介绍一下针对有限的内存，如果完成常规的计算。
