离线数仓

数仓采集

• 用户行为数据: 启动数据、页面数据、曝光数据、行为数据、错误数据。

埋点: 日志数据的产生需要通过用户触发埋点事件产生。

1. 代码埋点:通过调用埋点SDK函数,在需要埋点的业务逻辑功能位置调用接口,上报埋点数据。
2. 可视化埋点:需要集成采集SDK,不需要写代码,通过图形化界面的方式选出需要对用户行为进行捕捉的控件。
3. 全埋点:通过在产品中嵌入SDK,前端自动采集页面上的全部用户行为事件并上报埋点数据,相当于做了一个统一的埋点。

埋点数据上报时机:

1. 用户离开页面时:批量的处理当前页面发生的所有事件,减少了服务器接受数据的压力,但是不实时。
2. 每个动作、错误产生后立即发送: 实时发送用户的行为数据,但是服务器接受数据压力较大。

Note:采集到的日志中五种用户行为数据都聚集在一起的,因为采用了用户离开界面时上报埋点数据的方式, 当前页面的数据都是批次的上报,每次采集一批当前页面数据,不好将集中行为数据进行分开。

为什么选logback做为日志框架而不用log4j?

主要是选用了TaildirSource,因为TaildirSource是通过使用‘inode + 文件的绝对路径‘作为key来定位文件的位置信息(Value),而log4j生成日志的方式是按天滚动的,第二天会将之前生成的日志文件加上日期重命名,这会导致该文件的绝对路径发生变化,从而导致TaildirSource所定位的文件发生变化,所以我们选用 Logback作为日志框架,如果一定要用log4j作为日志框架,我们可以通过修改taildirsource源码仅通过inode来定位文件而不再依赖文件的绝对路径。

采集框架

Flume -> Flume (两种方案) 两层!

1. Taildir -> Memory(Kafka) Channel -> Kafka Sink | Kafka Source -> File(Memory) Channel -> HDFS Sink
2. Taildir -> Kafak Channel (生产者) | Kafka Channel(消费者) -> HDFS Sink

Note:

误区,通过一层Flume使用 TaildirSource + KafkaChannel + HDFS Sink的方式上传,表面上看似架构进行了优化,但本质上这个Flume会启动在LogerServer上,需要通过Taildir去采集数据,然后上传导Kafka,但Kafka跟LogerServer不在一个节点上,需要走网络IO将数据存储导Kafka,最后还需要 HDFS Sink将数据从Kafka在take到Flume所在节点即日志服务器,然后再通过这个Sink将数据写到HDFS,还是网络IO的方式进行传输,虽然只有一个Flume,但是一次日志的采集需要经过三次落盘,效率非常低,并且致命的是只采用一个Flume就需要每台日志服务器都要启动一个Flume去进行数据的采集,每个Flume都会有一个Sink,由于HDFS不支持并发写入,多个Sink需要向HDFS进行数据写入时会出现问题。

解释:没有Kafka Channel之前,有File Channel 和 Memory Channel。File基于磁盘数据要进行落盘处理,落盘会进行网络IO所以效率低;MemoryChannel基于内存无需将数据存储到磁盘,效率高,但是内存中的数据不安全可能会造成数据丢失,但是为了进行更高效的传输选择MemoryChannel。有KafkaChannel之后我们选择KafkaChannel,虽然Kafka也是基于磁盘的,但是数据的存储基于追加写入的方式也很快,并且使用KafkaChannel可以省去Sink组件,省去了take事务提高数据传输效率。通过Kafka进行数据传输时,对传输效率要求不那么高但是要求数据的安全性,所有选择File Channel。

为什么要第二个Flume?

第二个Flume完全可以不要,但是需要自己手动创建kafka消费者将Kafka中的数据消费出来,并且需要自己手动维护Offset,然后将消费到的数据再通过Hadoop的API上传到HDFS上,采用第二个Flume提供的kafkaSource和 HdfsSink可以代替我们完成这些操作,方便快捷,并且可以在第二个Flume中添加时间转移拦截器。

• 第一层Flume:使用taildir Source 解决了断点续传和实时监控的问题,使用kafka Channel直接对接kafka,没有sink组件,省去了take事务,提升了flume数据传输效率,并且将kafka Channel的参数parseAsEvent设置为false,解决了flume数据传入到kafka时,Json日志信息前面多拼接个header信息的问题,保证kafka里面的数据都是一个个的flume Body。

• 第一层拦截器:数据清理,对不满足要求的日志进行筛选过滤,空值、字段长度不够、非Json格式...

  作用:从数据源头对数据做ETL清洗，保证采集到的数据都是完整的json字符串，方便后续解析使用。

• 第二层FLume:使用kafkaSource对接kafka,获取kafka Topic的数据,传给file Channel,提高数据的安全性。最后利用hdfs Sink将数据传入到hdfs上。

  作用：可以定义上传到HDFS上文件的大小，设置滚动大小、滚动时间等。

• 第二层拦截器:两点,希望上传到HDFS上的一天的日志数据对应一个文件, HDFS Sink配置的HDFS路径中包含了时间转义序列,会默认从Event的Header读取timestamp,如果Header中有时间戳会使用Header中的时间戳,如果没有会使用本地时间戳。默认情况下Event中是没有timestamp这个header的,但是Kafka Source会用当前系统的时间戳作为timestamp的value放到header中,如果不通过拦截器去修改,HDFS上产生的文件会对应为本地系统时间,而不是当天日志时间。如果不添加拦截器,需要修改三台Kafka集群的本地时间。同时添加拦截器的另外一个原因就是为了保证当天产生的数据一定能在当天被采集到,23：59：59 产生的数据不会被因为采集延时被当作第二天的数据采集，可以解决数据漂移。这样一来我们就需要通过拦截器对Event中的Header进行处理,将日志中的时间戳作为Header的时间戳put到Event中。

说明：

第二种方案 Kafka Channel 对接 Kafka Channel的方式,通过kafka Channel直接获取kafka的数据,然后再用hdfs Sink传给HDFS。

好处:省去了Source组件,减少了put事务,提高了数据的传输效率。

缺点:由于第二个Flume需要一个拦截去对数据进行处理,使用这种方案第二个Flume没有Source组件,因此第二个Flume无法使用拦截器,只能将第二个Flume的拦截器挂载到第一个Flume的Source下,与ETL拦截器形成拦截器链,这会导致第一个Flume的Kafka Channel必须将parseAsEvent必须设置为true,数据进行Event格式解析,从而导致Kafka中的数据不再是单独的Json字符串,而会在前面拼接上一个header信息,这时只有第二个Flume的Kafka Channel也必须将parseAsEvent必须设置为true,才可以正确的解析Kafka日志中的数据。

数据采集完整性的检测:

通过 ‘wc -l‘指令查看log中生产数据的行数, 在通过hadoop ‘hadoop fs -text /hdfs路径‘ 在Linux上查看lzo压缩文件。对照两个文件的行数是否相同。

动态SQL:

后台的SQL需要通过前端传过来的过滤条件,参数个数不确定,为了匹配各种筛选情况,在SQL查询语句的最后添加where 1=1, 在其后面可以添加各种and筛选条件。

用户业务数据:

• 用户信息、商品信息、订单信息...都是以表的形式存储在数据库中。

同步策略

• 全量同步:每日存储一份完整数据作为一个分区,适用于数据量不大,且每日有新增及变化数据。
• 增量同步:每日存储一份增量数据作为一个分区,适用于数据量很大,且每日只有新增数据。
• 新增即变化同步:存储创建事件和操作事件都是今天的数据,适用于数据量很大,且每日有新增及变化数据。
• 特殊同步:地区维度、时间维度、民族维度等只需进行一次全量同步。

数据同步机制的实现:

新增遍历表的同步通过创建时间和修改时间为今天的作为筛选条件,新增及变化同步的表处理最复杂最棘手;增量同步通过操作时间是今天的作为筛选条件进行筛选。

数据仓库

• 范式理论:设计表的结构需要符合标准级别及规范要求,通过范式来降低数据的冗余。
• 函数依赖
  • 完全函数依赖:通过多个字段得到的值,但不能通过缺少任何一个字段而得到,则该字段完全依赖于其他字段。
  • 部分函数依赖:通过多个字段得到的值,可以通过缺少任何一个字段而得到,则该字段部分依赖于其他字段。
  • 传递函数依赖:通过某个字段得到的值,而该值又可以得到另外一个字段,则这两个字段为传递函数依赖。
• 第一范式:属性不可切分。
• 第二范式:不能存在部分函数依赖。
• 第三范式:不能存在传递函数依赖。

说明:范式等级越高,数据冗余越少,但是查询时会非常不方便,涉及到多张表进行Join,Hive中每join一次都会进行shuffle导致性能下降,而关系型数据库存在主键索引,每次join并不会严重影响查询效率。

数据建模

• 关系建模与维度建模:数据处理的两类方式。

1. OLTP联机事务处理:作为关系型数据库处理数据的方式,主要进行基本的日常事务处理,每次查询只返回少量记录,可以随机低延时的写入用户输入,数据表征为最新状态,每条数据的修改操作会覆盖之前的数据,数据规模为GB级别,MySQL最大容量约为500G。
2. OLAP联机分析处理:作为数据仓库系统的主要应用,支持复杂的分析操作,侧重决策支持,并且提供直观易懂的查询结果。对大量数据进行汇总,数据批量导入,数据表征为随历史时间变化,每次修改操作不会覆盖之前数据,比OLTP多维护了时间维度,数据规模为TB、PB级别。

说明：OLTP使用关系模型,关系模型就是ER,严格遵守三范式,因为当时磁盘比较贵,需要减少数据存储的冗余。OLAP需要使用维度模型。

维度建模

• 四步:选择业务过程 -> 声明粒度 -> 确认维度 -> 确认事实
• 维度模型：
  1. 星型模型:标准星型模型的维度只有一层。
  2. 雪花模型:雪花模型的维度涉及多层,较靠近3NF,但无法完全遵守。
  3. 星座模型:存在多个事实表的星型模型或雪花模型。
• 维度表与事实表
  1. 维度表:通常描述信息,作为描述事实表的一个角度,是一个宽表,与事实表相比行数较少,通常 <10 万条。 大部分维度表数据可控。
  2. 事实表:每行数据代表一个业务事件,每个事件都有相应的度量值,列数较少,是个窄表,行数很多数据量大,并且经常变化每日增加。 数据不可控。
     • 事务型事实表:事务一旦提交不再改变,更新方式为每日增量更新。(退单、订单状态、支付流水、订单详情、商品评论、活动参与订单关系表)
     • 周期型快照事实表:不会保留所有数据,只保留固定时间间隔的数据,更新方式为每日全量更新,即每日对数据做一个快照。(除省份、地区之外的所有表)
     • 累积型快照事实表:累积快照事实表用于跟踪业务事实的变化,业务事实不能一次完成会随时间而改变,更新方式为每日新增及变化。(优惠卷领用表、用户表、订单表)

数仓分层

• 分层的作用:

  <1>复杂问题逻辑的简化,减少重复计算,增加单次计算的复用性。

  <2>隔离原始数据,方便权限管理,与敏感数据与异常数据解耦。

• ODS:

  1. 原始数据层,保持数据原貌不做任何修改,起到备份数据的作用。
  2. 数据采用压缩,减少存储空间。通常压缩后文件大小会到原来的1/10。
  3. 创建分区表,防止全表扫描。

  表结构:

  日志数据:只有一个字段,是一条Json数据。

  业务数据:对应发SQL表中的字段,针对不同的表采用不同的同步方式。

  说明：只有ODS层才有索引和同步策略,其他层不使用同步策略,只需将上层数导入即可。

• DWD:

  • 数据清洗:

    1. 去除json数据中的废弃字段

    2. 过滤日志中缺少关键字段的记录

    3. 过滤日志中不符合时间段的记录

       方式：Map、SparkSQL、Kettle

  • 数据脱敏:

    对用户的敏感信息进行脱敏隐藏,姓名、手机号、身份证号、家庭住址等。

  • 维度退化：

    采用星型模型构成的星座模型,单个事实表外层只有一层维度表。

  • 维度建模:

    1. 选择业务过程 -> 根据需求选择事实表,每条业务线对应一个事实表。

    2. 声明粒度 -> 精确定义事实表中的一行数据代表什么,需要选择最小粒度,这样所有的合理需求都可以从DWD层获取,如果聚合后的函数则无法获取明细。

       数据粒度:指数仓中保存数据的细化程度或综合程度级别。

    3. 确认维度 -> 通过不同的维度描述业务事实,即描述事实的角度。

    4. 确认事实 -> 事务的度量值,通过维度+度量值来描述事实。

    *该层都是数据明细,数据量大查询效率低,并且存在重复计算问题。

    *缓慢变化维:数据会发生变化,变化频率也不高,但大部分数据不会发生变化,数据缓慢变化,解决缓慢变化为常用的方式使用拉链表。

    拉链表:拉链表的初始时间需要获取数据的全量表,并初始化开始时间和结束时间。

• DWS:

  • 将DWD层的多张表进行Join形成宽表。
  • 统计各个主题对象当天行为,服务于DWT层的主题宽表。
  • DWS层以维度为基准,去关联与该维度对应的事实表,然后关注聚合后的度量值。
  • 提高数据大量查询时的效率,并增加一次计算结果的复用性。
  • 主题宽表会用所有与这个主题相关的事实表度量值的统计结果作为主题表的字段。

• DWT:

  • 统计主题对象的累计行为,与DWS层一样以维度为基准，去关联对应多个事实表。
  • 在维度表的角度去看事实表，重点关注事实表度量值的累积值、事实表行为的首次和末次时间。

• ADS:

  根据各种需求建不同的表。

说明： DWD层是以业务过程为驱动,DWS层、DWT层和ADS层都是以需求为驱动。

数仓搭建注意事项

• 日志的解析:通过get_json_object函数来获取Json中的内部数据。
  • 启动日志:启动日志只有一个Json,将Json中的字段导入到DWD层对应表中的字段即可。
  • 事件日志:事件日志不是一个标注的Json,内部有一个Json数组et,这个数组中的事件是随用户行为而改变的无法预估,针对于内部的Json数组,我们采用自定义UDTF函数的方式将其分解开配合获取内部数据。
• 数据的插入方式
  • Load:通过外部文件直接将数据导入表中,通过外部导入文件添加数据的方式不能使用列式存储。
  • Select Insert:通过查询其他表将结果插入到目标表的方式。
• 数据存储格式:
  • 行式存储:TextFile、SequenceFile(二进制)
  • 列式存储:ORC、Parquet
    • 除ODS层和ADS层外,采用列式存储并且每列采用内压缩,即内部每个列都压缩了,整体还是以外部的存储格式为准。
    • ODS层通过加载外部文件的方式添加数据,不能采用列式存储,采用TextFile存储和LZO外压缩,并对LZO创建索引。
    • ADS层数据量比较小且ADS层主要用来做查询,不采用列式存储,也不采用压缩。
    • 采用列式存储的表只能采用Select+Insert+表 的方式添加数据,不能通过Load的方式添加数据。
    • 数仓项目采用parquet存储+LZO压缩,是因为比较快,且支持切片,但是本身可以不用进行切片,因为数据采集导入hdfs是采用的hdfs Sink,设置了上传文件的大小默认是128,每一个文件根本不可能超过128,所以不用切片。
• Hive 数据数据的格式:
• Hive中默认的输入数据格式为‘CombineHiveInputformat‘,用户可以在创建表时在存储格式中指明输入输出格式,但只有ODS层才有索引,是外部文件输入,读取表中数据时需要设置输入数据格式。
• Fetch抓取:Select ‘*‘ 时不走MR任务,会走Fetch抓取模式,读取数据时会按照创建表时指定的数据输入格式。
• MR任务:Select ‘count(*)‘时会走MR任务,会按照Hive默认的输入数据格式即‘CombineHiveInputformat‘格式,无法识别LZO文件,因此需要通过‘SET hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;‘设置Hive的数据输入格式。
• Hive 数据类型:
  • 向Hive表中添加字符串类型的数据时,注意类型‘string‘,本身就是小写。
• Hive 函数说明:
  • Concat_ws只能填字符串或字符串数组、Hive只能识别的时间格式连接符为‘-‘,其他格式需要使用‘regex_replace‘进行替换。、
• Sqoop:
  • 支持通过sql进行行列过滤导入并且可以导入hdfs、Hive、HBase。
  • 导出只能从hdfs往外导,不支持行列过滤只能进行全表扫描。存在数据重复问题。
  • 底层跑的也是rm任务,只有map没有reduce,默认4个map。
• Azkaban:
  • 底层三部分 Webserver、Executor、Mysql。

业务SQL注意事项

• 用户表:数据存在T+1 跨天问题。
• 订单详情表:采用每日新增及变化同步策略,不用考虑跨天支付的情况。
• 金额分摊:当数据不是整数时存在精度损失问题,当大量的损失聚合时很大的损失,需要求分摊金额总和并将损失金额精度找出并补回。
• DWT层获取用户数存在时效性问题,新旧对比时新的活跃数据会覆盖掉旧的活跃数据末次登录日期。有些数据可以从dws层直接获取,每天一个分区,分区内不重复但分区间可能存在重复,多个分区进行合并时记得去重。
• 会员新鲜度:新增用户/活跃用户。
• 沉默用户:首次登录时间=末次登录时间,且注册时间在7天前。
• GMV:无活动正常 300-500,有活动时能有800-1000。
• DAU:正常百万日活。

Offline DW

原文：https://www.cnblogs.com/yuexiuping/p/14956634.html