基于大数据开发平台（Hadoop）和基本的大数据中间件（spark，hue，hive，flink等）集成出来的能处理公司日常大数据需求的一个整体技术框架。

关于大数据架构的方案，目前没有一套标准的方案，也没有一套体系能适用所有的公司，每个公司都是结合自己的业务场景，数据规模以及对应的数据需求来选择适合自己公司的底层框架和大数据中间件产品。

基础的架构主要是基于已有的大数据平台和中间件的开发，将一个公司已有的数据或者外部接入的数据进行存储和聚合以及能对已有的数据进行分析，得出对业务支撑的一些数据指标。

基本架构图：

数据源的采集

1、任何大数据的框架都是依赖于自己所有的数据源，一般的数据源分类为：

自己业务数据（主要是各个业务线的数据，例如：用户、订单等），主要存放在关系型数据库中，例如Mysql等。
埋点日志（涉及到用户对APP或者小程序等的访问、分享、浏览等基本的用户行为数据），这份数据主要是存放在HIVE上；
第三方数据，例如：获取的第三方竞品数据、各种API获取、TD投放、爬虫等第三方数据。

2、数据的采集方式：

业务数据的采集：主要是做实时的数据管控以及和其他数据的整合。

业务数据一般存放在自己内部的mysql数据库里面，里面涉及到公司的私密数据和重要数据，其中大部分的数据是不存历史快照的，只有当前状态的更新。

业务数据的采集一般的流程：

业务库的数据进行binlog监听，获取线上业务的数据。

一般常见的监听架构为：

Mysql+Canal+Kafka+Streaming/Flink

2.埋点日志的采集：主要是离线数仓和实时采集

目前我们的大数据架构分层两块：

1、NG日志 -> Flume -> logrotate每10分钟抽取一次 -> HDFS -> Hive

2、NG日志 -> Kafka -> ES/Redis/mysql

不过第一种方案是又一个数据丢失的风险，在logrotate每10分钟抽取一次的情况下，会出现文件轮转的时候大概100ms的丢失，因为轮转操作是先将Flume生成的日志文件MV到一个新的文件中，然后在重新建一个新的日志文件，这个过程中，Flume日志依旧会在写数据，但是文件没有，导致数据丢失。

所以目前的改进方案，离线和实时一起做：

NG -> Kafka -> StructuredStreaming/Flink

然后实时的需求直接走StructuredStreaming/Flink，获取实时的数据存到redis/ES等内存数据库中，可以做搜索推荐。

离线的将数据存到HDFS中，第二天对昨天的日志进行合并（主要是合并小文件）。

hadoop fs -getmerge /*/*  /*/*

原理：

StructuredStreaming/Flink主要的模块都是两个：Source+Transform+Sink

其中Source可以支持File Source/Kafka Source等，Transform主要是对数据源的数据处理和转换，主要是DataFrame的转换（尽量是一些简单的操作），Sink主要是FileSink/KafkaSink/ForeachSink。其中写到HDFS中使用FileSink，其他的比如写到mysql、ES、Redis都可以使用ForeachSink。

例如：

RedisSInk 
class redisSink extends ForeachWriter[Row](){
  override def open(partitionId: Long, version: Long): Boolean ={
     //这个方法进行一些初始化，如redis，获取连接
  }
  override def process(value: Row): Unit ={
    //具体的处理逻辑，写数据到数据库中
  }
  override def close(errorOrNull: Throwable): Unit = {
   //关闭连接
    }
}

例如：
class MysqlSink(url: String, user: String, pwd: String) extends ForeachWriter[Row] {
    var conn: Connection = _


    override def open(partitionId: Long, epochId: Long): Boolean = {
      Class.forName("com.mysql.jdbc.Driver")
      conn = DriverManager.getConnection(url, user, pwd)
      true
    }


    override def process(value: Row): Unit = {
      val p = conn.prepareStatement("replace into table(seller_id,pv) values(?,?)")
      p.setString(1, value().toString)
      p.setLong(2, value(1).toString.toLong)
      p.execute()
    }


    override def close(errorOrNull: Throwable): Unit = {
      conn.close()
    }
  }

3.第三方日志采集

第三方日志主要是自己投放日志、爬虫日志或者投放日志。这块主要是调用第三方API接口，对数据进行回调，然后和上面的两份数据进行糅合，生产各类的业务数据，例如：用户画像、投放率、转化率、用户分层等。

数据中台

数据中台是基于业务来创建的一个聚合和治理跨域数据，将数据抽象封装成服务，提供给前台以业务价值的逻辑概念。

业务痛点：

面向业务的服务能力不足，业务需求响应速度慢；
数据口径不一致，指标定义不明确，导致数据准确性不高；
开发流程比较繁琐，资源浪费，数据流程不清晰；
业务驱动数据，而不是数据驱动业务，被动做需求；
数据复用性不高，一个业务一套数据，数据冗余；

目标：快速的提供业务支撑能力；

业务类型：复杂的数据模型，统一指标体系的数据产出；

输出方式：提供数据API服务；

底层逻辑：对同一业务线的数据进行聚合，不同业务线数据低耦合，对共性的题抽象出业务的规则，建立数据模型，提供高复用的数据资产。

数据中台最主要模块

数据治理：数据治理是将大量数据（采集过来的数据）进行整合，清洗重复的数据，确定唯一性指标，做好血缘关系，产出规范的数据模型，保障数据指标的可信度。

主要是数据字典（方便数据查询）、数据指标（确保指标的唯一性）、血缘关系（确定数据依赖关系）。

数据安全：除去直接的业务支撑数据，其他沉淀的数据是按照业务线进行分层存储，对外封装服务，可以根据业务线进行权限管控。多个域的数据进行聚合后生成的数据宽表，需要通过获取字段的权限来进行管控。

平台权限管控、数据权限管控、队列管控。

数据维护：

调度：数据的基本执行调度都是依赖公司内部平台，内部需要建立一个自己任务的校验机制，判断每天结果表的条数阈值，来判断依赖的数仓数据是否执行完，并设置循环调度；

数据准确性：依赖数仓，确定重要指标的主表，建立各个指标的阈值范围，监控数据表重要数据指标的波动情况，建立血缘关系，一旦出现大范围浮动，快速定位，查询问题。

数据服务

主要是接口服务和可视化服务。

可视化报表主要是数据管理后台、数据大屏等，可自建也可使用第三方开源工具，例如：matebase、QuickBI、GrowingIO、神策等。

服务规范：

第一要务：完成复杂业务规则的数据支撑，直接对接业务；

在做业务的基础上，了解不通业务线的流程，获取该业务线不同类型的数据进行聚合，生成不通维度的基本数据。在利用这些数据进行类似于画像、分层、健康度等专题分析；

第二：不同业务线上抽象出一些共性的指标，建立一个高复用的模型，可以通过不同业务线上指标数据的对比来对业务进行指导；

第三：将完善的、已被验证过的数据封装成接口，对外开放，不同业务可以直接获取想要的数据；

ITPub博客

大数据开发架构

数据源的采集

数据中台

数据服务

最新文章