基于流式计算的数据抽取系统设计

王 岩

北京搜房网络技术有限公司

基于流式计算的数据抽取系统设计

王 岩

北京搜房网络技术有限公司

在现在这个互联网飞速发展的时代，企业数据也成几何级增长，如何从海量数据中提取、整理出企业想要的数据，使看似无用的数据变得有价值，数据抽取也许是开发过程中经常遇到的问题。通过使用流式计算的思想，对原有系统数据源进行分类，引入消息队列、数据缓冲池、并行消息队列等概念，可以很好的解决此问题。

消息队列、数据缓冲池、流式运算、生产者消费者

1、传统数据抽取系统设计介绍

面对上百万的数据需要清理、加工成新的数据，并且在清洗数据的过程中还要与其他数据库、第三方Web服务进行交互，如果按照传统的设计思路，我们经常会采用分页方式及顺序执行的方式进行设计。如：通过分页方式提取数据，然后从第一页到最后一页数据进行与其他数据库进行数据清理、加工，之后再把整理好的数据以分页的方式从第一页到最后一页与第三方Web服务进行交互，最后在用分页方式将所有数据存入到数据库中。这种方式是一种开发高效的设计方式，可以在一定程度上节省开发时间和成本，但却是一个紧绷的结构，通常会造成程序运行时间的浪费，而且由于步骤上下游的耦合度过高，经常会遇到上游宕机，下游无法处理等问题。虽然看似这种设计方式可以节约开发成本，但是因为整体结构设计紧耦，会导致项目后期维护难度增加。

2、流式计算的数据抽取系统设计

流运算是计算机程序的一种数据整理、分析的方式，在流运算当中，高级软件的运算法则在接收流数据时就开始对其进行分析、整理。了解了流运算的概念之后，接下来我们所要做的就是对数据、步骤进行分类、业务解耦。

2.1 数据分类

基础数据源，我们将被提取的数据称之为基础数据源，基础数据源可以按特征进行数据分类，如：城市、日期等。注意：选择的特征分类不易过多，且特征可以涵盖所有数据源信息。

数据字典，是指将一些数据字典或者数据量小且不变的数据称之为数据字典，这种数据数据量小，但在清洗、加工数据过程中起关键性作用，如：城市与城市简称对应表、ID与相应名称对应表等。

目的数据，我们将最终得到的数据称之为目的数据，它经过多个步骤最终存入数据库或者形成XML服务。

第三方数据源，因为基础数据源无法直接作为最终的目的数据，需要通过其他数据库进行数据整理、加工，我们称之为第三方数据源，这种数据源在整个数据整理的过程中，有可能会有多次并且是不同数据库的第三方数据源。

第三方Web服务，通过基础数据源的某一列的值进行Http请求获得相应数据，称之为第三方Web服务。

2.2 流式运算设计

流式运算设计的主要思想是解耦和步骤的拆分，数据的分类其实是对业务的整理和解耦，并且按照不同类型的数据操作进行步骤的拆分。流式运算所采用的核心技术是消息队列。

消息队列中间件是大型系统中不可缺少的组件，它采用数据先进先出的原则，解决系统逻辑耦合，异步消息，流量削锋等问题，实现高性能，高可用，可伸缩的架构。常见的消息队列中间件有：MSMQ、Kafka、RabitMQ等。消息队列常见的模型有：点对点和发布、订阅。消息队列核心设计思想是生产者、消费者模式，在实际的软件开发过程中，经常会碰到如下场景：某个模块负责产生数据，这些数据由另一个模块来负责处理(此处模块是广义的，可以是类、函数、线程等)。产生数据的模块，就形象地称为生产者；而处理数据的模块，就称为消费者。消息队列具有可靠缓存，可以防止数据丢失，防止数据堆积；消息队列还具有流式模型，因为它的生产者、消费者模型也完全符合这种storm(流)模式；消息队列还具有消息总线功能，它以一种总线的方式出现，负责消息的传递和分发，直接减少了模块间的耦合程度。

对于原有系统的改造中，因为业务数据的多样性和复杂性，我们需要使用多个消息队列组合使用，将不同的数据分类固化到不同的消息队列当中，如：第一个消息队列提取全部的数据字典和分页方式提取基础数据源，第二个消息队列与第三方数据进行交互处理数据，第三个消息队列与Web服务进行交互处理数据等等。同时，为了更好的提升每个步骤的处理效率，我们引入了数据缓冲池和并行消息队列的概念。数据缓冲池是消息队列与消息队列之前通过程序运用缓存技术构建的一个数据缓冲池，当数据缓冲池达到一定数量的时候开启下一个消息队列进行数据运输、处理，例如：如果数据缓冲池阈值设置为100，那么如果当某个消息队列每次处理20条数据式，这个消息队列必须运行5次后，使得数据缓冲池大小达到100以后，才能开启它后面的消息队列。并行消息队列的概念是指每个消息队列可以横向扩充，例如：当数据缓冲池中有100条数据时，每条消息队列处理20条数据，为了提高运行速度，可以通过代码增加了5条消息队列，那么当程序运行时候5条消息队列并行处理，这5条消息队列称之为并行消息队列。注意，并行消息队列采用多线程开发的思想，并行消息队列越多会对这个消息队列的第三方数据库或者第三方Web服务造成负载压力，配置不合理反而会造成性能问题。

3 流式运算架构设计（见图1）

在第一个消息队列启动之前，数据字典通过全量的方式，加载到程序缓存当中。因为基础数据源通过不同特征(如：城市)进行了分类，程序通过多线程的方式，开启不同线程执行各自的程序，即北京、上海、天津等城市并行执行后续消息队列操作，以下举例为分类为北京的数据，第一个消息队列(Queue1)开始以分页的方式对基础数据源进行抽取。当第一个数据缓冲池满了以后，开启第二个消息队列(Queue2)，程序与第三方数据源进行交互并对数据进行加工。当第二个数据缓冲池满了以后开启第三个消息队列(Queue3)，并通过程序请求第三方Web服务对数据进行加工。最后当第三个数据缓冲池满了以后，开启第四个消息队列，第四个队列(Queue4)对最后形成目的数据，并入库或者形成xml服务。如果过程中还有其它第三方数据源或者第三方Web服务，可以纵向增加消息队列(QueueN)即增加了数据处理步骤。如果当某个消息队列时间消耗过长或者阻塞时候，可以通过增加并行消息队列，提高并发，例如：Queue31、Queue32。在流式运算数据抽取系统中，需要在程序中增加运行日志，运行日志主要包括每个消息队列的开始时间和结束时间，每个步骤中出错日志，以及处理多少条过滤多少条，因为流式运算系统，运行日志起到对整体程序健康度的监测，通过监测每个消息队列的运行日志中的运行时间，可以在耗时较长的消息队列增加并行消息队列来提高整体的运算速度。

图1

这种流式运算数据抽取系统的设计，实现程序的纵向拓展和横向拓展，通过增加消息队列实现纵向拓展，使数据处理之间解耦。通过增加并行消息队列实现横向拓展，提高单步并发。在实际应用中这种设计方式，可以大大提高程序每分钟处理量，降低程序之间的耦合性，并且不会出现某一部分数据出现问题导致整个程序无法产出目的数据。在开发方面，流式运算方式将步骤解耦，搭建好整体框架以后可以很好的分配相关人员开发相关业务，大大提高了开发质量和开发效率。

4、结束语

流式运算是处理大数据手段之一，通过多个消息队列可以搭建一个强壮的流式运算框架，再根据需求开发不同业务，就形成了流式计算的数据抽取系统，这种系统属于后台系统，多与系统中的计划任务结合使用，通过计划任务定时定点启动流式计算数据抽取系统，实现数据的筛选整理，入库或产出相关的数据服务，为后续业务系统提供基础数据源进行分析展示。