小米有众多的智能终端和设备，数据规模非常大，对于数据采集和大数据集成提出了非常高的要求。此次演讲主要介绍小米大数据集成解决方案，主要包括小米数据流平台的架构演化，整个链路的数据质量监控，数据流生态的构建思路，最后会介绍典型的应用场景、未来的规划和思考。
分享大纲：
1、问题与挑战
2、数据流整体框架
3、核心功能
4、应用场景解析
正文：
1、问题与挑战
首先，我介绍一下小米大数据集成架构面临的问题和挑战：一是大数据场景下系统众多，包括各种存储系统和计算系统，这其中很关键的一个问题是如何高效集成所有系统以让数据发挥最大价值;二是做数据集成时，我们希望可以有更低延迟;三是当数据在不同系统中流动时，我们如何及时发现并解决问题;四是量化数据在整个链路传输过程中存在的问题，比如数据延迟、数据丢失等。
在这里我还是要推荐下我自己建的大数据学习交流qq裙：458345782， 裙 里都是学大数据开发的，如果你正在学习大数据 ，小编欢迎你加入，大家都是软件开发党，不定期分享干货（只有大数据开发相关的），包括我自己整理的一份最新的大数据进阶资料和高级开发教程，欢迎进阶中和进想深入大数据的小伙伴。
2、数据流整体框架
上图为小米数据流整体架构，大致可分为三部分：中间层叫Talos，这是小米自研的一套消息队列，其主要应用场景有两个：一是作为数据中间件(数据中转中心)，二是服务于后续的流式计算。虽然现在Kafka非常火，但是我们确实发现了Kafka的一些问题，比如Reblance、扩容、缩容等问题，因此我们选择使用自研Talos。下层是基于流式消息队列做的source和sink扩展，目标是希望以Talos为数据总线把大数据应用场景下的不同平台连接起来。上层依赖于底层的source和sink体系，解决Metric监控、报警和数据收集等问题，也会做OLAP分析和线上APP日志收集等。我们希望在这个架构下，业务方可以根据不同需求适配该系统以得到完美的解决方案。
3、核心功能
上图为该平台的主要核心模块，最底层是消息队列，中间层是数据接入层，该层包含一些SDK，因为它本身作为消息队列也存在很多应用场景，比如推送场景等。我们在消息队列上做了Streaming Plunigs以适配各种Streaming系统，其后的Source和Sink其实是对数据的扩展。最上层主要是基于这套框架做的整体feature，比如全局web端控制和产品化方案。我们也做了全链路数据监控和数据追踪。我们有一个自己的流式计算管理平台，用来帮助管理用户流式作业。因为小米有自己的海外业务，所以我们有全局数据中心的数据replication机制，需要在海外部署自己的数据中心，因为数据是全球分散化的，必然就存在数据汇总问题，这些构成了我们的整体解决方案。
接下来，我将逐步介绍核心功能。首先，我介绍一般数据解决方案(如上图)，这是一些常见系统，我们要解决的问题是在不同的系统间做数据集成。
这之中存在一些问题：一是系统的交互复杂度较高;二是当涉及的系统较多时，如果每个业务都做自己的事情，不仅前期研发成本会非常高，而且后续功能添加、系统运维、系统重构和交接成本会更高;三是当各业务方按照自身需求进行开发时，由于彼此独立导致无法复用重复逻辑;四是如果让业务方完成某件事情，业务方往往会忽略监控和数据质量，一般缺失或者很难做到很完备，因此无法保证数据交互质量;五是一般由业务独立部署，基本无法抽象化和服务化，进而很难积累经验和传承知识。
我们的做法是基于Talos消息队列作为消息总线，和不同的系统进行交互，我们在外围做一些产品化工作，接管一些比较繁琐的配置。基于此，我们做了Multi Source和Multi Sink的抽象。
我们认为抽象首先是希望数据在系统间进行流动，这之中包含两层含义：一是连接不同的系统;二是构建低延迟场景。我们做产品化封装其实就是避免业务团队的重复投入与研发。我们希望以Talos为消息数据总线对所有数据使用流式计算以尽可能降低数据结构产生的延迟。要想在所有系统之间作中转，我们就需要考虑集成复杂度，通过Source和Sink组合的模式，系统集成复杂度降为O(N)。最后，我们期望按照这种模式接入更多系统，这样我们就可以形成规模效应，依赖目前的数据流平台给用户产生价值，建立数据流生态系统。
接下来，我们介绍一下系统监控。对于整套流程，我认为系统监控大概分为以下几方面：数据丢失监控，数据延迟增加监控，服务进程异常监控，流量异常监控。
一般情况下，我们会在每台机器上部署一个agent，收集不同模块的Metric数据。其次，我们会把数据汇总到消息队列，对数据进行各种整理并中转到Druid平台，由于监控数据有很多维度，因此我们用Druid的目的就是降维，按照不同维度进行数据合并。然后，我们将数据中转到Falcon，这是一套小米开源的监控系统，我们会对监控数据提供Web化展示，让用户实时看到监控内容，我们也会周期性生成报表来告诉用户一些数据情况。
说到底，数据流端到端审计其实就是量化整个链路的数据情况，其展现形式就是Web界面，能够实时查询最新数据，也能查询历史数据。在报表部分，我们引入了两个概念：Event Time和 Processing Time，Event Time指消息真正产生的系统时间。Processing Time指消息到达具体某个模块，因为我们是一个流动的系统，里面有很多模块，我们会把 Event Time和 Processing Time依赖于类似Metric的模块进行数据打点并收集。我们会把Event Time看作消息的唯一标签来进行数据校验。Processing Time用来统计消息从产生到达某一模块的系统延迟情况。
上图为数据流端到端审计的简单处理流程，我们会在每一个节点做埋点数据，埋点数据主要包括每条消息的Event Time和 Processing Time，以及其归属的流等类似信息。为了保证监控数据的高可用，我们会把监控数据先持久化到本地磁盘。然后，我们通过一套自己的流程收集数据。
整个过程存在几个问题，一是监控数据量非常大，如果要监控整个链路，我们需要给每条流经系统的数据做埋点，这会导致系统负载增加;二是消息有可能重复，在最终结果汇总时，我们要做相应的去重处理;三是期望数据以一种准实时的形式展示给用户。
在这之中，我们引入了Spark Streaming对数据进行处理，虽然消息有重复，虽然Spark Streaming作业会因为各种原因挂掉，但我们需要保证消息正确统计。我们在Agent端进行分钟级别的数据合并，也就是在每台机器上进行预聚合，这样能够把原始监控消息的数据量大大降低。我们会在Spark Streaming里做一些基于内存的去重逻辑，依赖外部KV系统做数据校验。
上图为我们的监控成果，小米会有很多线上数据，这些数据在进行埋点后写入消息队列，最后会转到Kudu和HDFS中，HDFS后期会做一些基于离线的Hive分析过渡，或者是OLAP分析。上图展示了我们到达这个环节的数据量，对于一个时间窗口，也就是一分钟的时间内，窗口所到达的数据量。
上图显示出现了一些数据丢失。目前来看，整个过程应该是数据从线上灌入Hive和HDFS，上图所示数据为测试样例，展示了从消息队列到后端部分系统做转储的过程。
4、应用场景解析
因为篇幅有限，我就介绍了小米对Source和Sink理念的理解，以及我们如何做监控和数据审计。我认为，对大数据集成系统而言，这三点对对业务来说是至关重要的。接下来，我将解析部分应用场景。
首先介绍小米内部数据集成的典型方案——埋点数据收集。对互联网公司而言，埋点数据收集是非常重要的应用场景。上图大概分为几部分，一是对各种web服务设置埋点数据，通过扩展appender来实现业务，方便的把数据通过每台机器部署的agent进行中转;其次，我们也做了web接入层，将大量的离散点通过这种模式收集起来，进而对其做数据分析。
第二个应用场景是实时日志分析，对于线上日志文件，我们会通过agent进行监控，将数据实时传入Talos，通过ES或者Kibana进行实时查询。简单来说，我们依赖整个系统把分布式文件通过准实时的方式导入ES，再通过Kibana进行查询和可视化日志分析。
第三个场景是泛OLAP场景。我们通过Druid做多维度分析，使用Kudu进行即即席询，利用Kudu的列存储以较低延迟展示数据， 利用Kylin做T+1查询。
整体流程如上图所示，小米目前还存在大量MySQL集群和报表数据，MySQL可以满足OTLP的需求，但是对于很多的OLAP查询，延迟会比较大，无法满足需求。基于此场景，我们的解决方案是把MySQL数据以准实时方式导出，在Kudu里做一个实时镜像，通过Spark SQL做查询。或者，将HDFS的数据灌入kudu，再接入Superset之类做展示。
最后一个场景是流式计算。我们之前提到系统中有很多Source和Sink，我们期望能够从Talos也就是消息队列里将数据传入Spark Streaming做实时计算，综上，这就是小米在数据流系统构建方面的经验。
在这里我还是要推荐下我自己建的大数据学习交流qq裙：458345782， 裙 里都是学大数据开发的，如果你正在学习大数据 ，小编欢迎你加入，大家都是软件开发党，不定期分享干货（只有大数据开发相关的），包括我自己整理的一份最新的大数据进阶资料和高级开发教程，欢迎进阶中和进想深入大数据的小伙伴。
