导语：穿梭机是致诚与宜信基础研发部合作开发的包含实时清洗、衍生变量计算、和数据融合入库等功能的一体化系统。
一、简介
一般而言，大数据处理流程主要包括数据采集、数据处理、数据存储、数据挖掘和数据应用等环节，大数据处理流程每个环节的数据质量最终决定了大数据产品的质量。相较于传统的软件，大数据产品要有大量的数据规模、快速的数据处理、精确的数据分析等特点。
“穿梭机”便是用于衔接大规模的数据采集和海量的数据存储，并在其中实现快速的数据处理的重要枢纽。本文将结合大数据处理流程介绍穿梭机在实践当中的演进，正如穿梭机这个名字，我们希望繁杂琐碎的数据碎片，如同穿梭时空一样，以一个干净有价值的新姿态呈现到我们面前。
从大数据处理流程的第一个环节开始。数据采集，就是通过各种方式收集数据，比如异构数据库、日志、用户事件和公开信息等，是大数据产生的源头。异构数据的迁移有Apache Sqoop等批量数据传输组件，日志收集有Logstash等日志收集组件，公开信息采集有各种爬虫组件。
由于采集数据的方式不同，数据又可分为结构化数据和非结构化数据。比如来自Oracle、MySQL的数据，对字段和结构有着严格的定义，因此他们是结构化数据。而来自MongoDB、ElasticSearch的数据，通常字段不统一，结构不相同，是非结构化数据。同样，日志和采集的公开信息，他们也是非结构化数据。
公开信息采集是我们重要的大数据来源，其更贴近互联网的一侧，我们利用公有云来搭建数据采集中心。在系统演进过程中，采集中心已经具备了任务分级、弹性收缩、实时监控等的功能，可以合理支配资源、分配任务，使得数据采集在有限的硬件条件下尽可能高效。
采集完数据，第二步是对数据进行处理。由于原始数据产生自外部网站、服务接口等不确定的数据源，往往会出现数据缺失、错位或错误等问题，在存储之前，会进行数据预处理，数据预处理一般包含过滤丢弃、去除重复、归一化、修正错误、结构化等数据清洗工作。在这之后，还会根据业务特点和数据种类，做进一步的处理，从而提高数据的价值性、可用性、时效性和准确性。
数据处理完成之后，根据数据的用途和结构，将其写入合适的存储。比如用于模糊搜索的数据，将其写到ElasticSearch。对于格式复杂、涉及多层嵌套的数据，将其写入MongoDB。对于大体积很少修改的数据，将其写入HDFS。
对于数据分析和数据应用，由于不属于穿梭机的工作范围，本文不再展开阐述。
三、穿梭机的前世今生
云环境资源有限，计算能力较低，大量的计算更适合于放在本地环境，但受限于云环境和本地环境的先天隔绝，数据传输的时效性和数据计算的准确性是一直以来最受关注的点。
穿梭机项目诞生之初，我们在云上搭建了临时的中转数据库，采集到的数据先实时存储在中转数据库中，次日凌晨再将前一天的数据批量拉回本地数据库，然后在本地对前一天的数据进行离线计算，最后将离线计算的结果分发到其他存储。
这是一条比较长的链路，每个环节也无法保证自身的高可用。数据保存在中转数据库时，数据本身已经丢失了一定的时效性。并且，由于次日才对数据做处理，很难第一时间发现数据源头的问题，故障响应时间也因此延长。
随着系统的演进和Wormhole的加入（Wormhole 是一个一站式流式处理云平台解决方案，详见edp963/wormhole），业务中的痛点被逐渐的解决，时效性和准确性得到了很高的提升。
现在，我们将采集下来的数据实时发送到消息队列Kafka，Wormhole以Kafka为Source端，对数据进行统一的处理，处理完成后，由Wormhole写到各个Sink端。
从高可用上分析，依托于Zookeeper，Kafka集群本身实现了高可用，确保其作为数据管道，可以可靠地传输数据。Wormhole，作为运行在YARN集群上的Application，同样保证了高可用。移除了本地中心库，数据直接写入业务相关存储，消除了单点故障导致的系统风险，所以目前数据处理的可靠性得到了保证。
从时效上分析，采集到的数据实时发送到Kafka，延迟在毫秒级，Wormhole实时处理写入存储，延迟在秒级。所以现在的架构，从数据采集到存储，实现了秒级延迟。从数据的计算方式来说，原来数据是离线计算，时效是T 1甚至是T N，现在改为实时计算，时效提升为T 0，或者可以说是秒级延迟。
从架构上分析，这是一个Kappa架构。通过Kafka来保存近期的数据，每类数据保存在不同的Topic中，根据不同的业务需求来精细地控制每类数据的保留时间。当需要对某种数据重新计算或者新的计算时，就可以通过控制Flow消费数据的位置来实现，然后将结果保存在新的存储当中。在这种架构下，我们完成了所有数据的数据清洗、数据融合和衍生变量计算等工作。
由于采集的数据多为非结构化的数据，用Wormhole实现了数据结构化。我们使用了UMS_Extension来定义数据的格式和类型，数据在真正处理之前，经过了一次字段和类型的统一，这减轻了数据处理的复杂程度。
在数据的处理过程中，由于数据清洗过程的复杂，单一的SQL难以处理所有的情况，我们大量使用Custom Class Transformation。为此，我们开发了Custom Class框架和单元测试框架。
通过不同的继承即可实现不同级别数据结构的转换，使得开发人员更能专注于对数据本身的处理。
有了很多Custom Class Transformation类，我们就需要简单的方式对代码进行单元测试。为此，我们基于TestNG开发了Transformation单元测试框架，使用注解即可进行方便的测试，保证了数据处理的准确性。
最后，我们使用非关系型数据库来保存处理好的数据，如ElasticSearch和MongoDB。使用Wormhole的Sink Schema功能，在数据输出之前再对字段和类型做一次统一，并且支持输出类似JSON的复杂类型，把好数据处理的最后一道关。
至此，介绍完了致诚穿梭机项目的前世今生。
写在最后：目前，得益于Wormhole实时流式处理方式，数据处理的压力被分散开来，机器资源消耗峰值得以降低。在3个节点的集群中（单节点Xeon E5-2680，2.40GHz，64GB），在CPU使用率低于50%，内存使用率低于60%的情况下，穿梭机每天承担着千万级的数据处理，数十GB的数据累计。
,
