CN105205687A

CN105205687A - 一种海量数据采集的方法

Info

Publication number: CN105205687A
Application number: CN201510521135.7A
Authority: CN
Inventors: 商希森; 付传伟
Original assignee: Inspur General Software Co Ltd
Current assignee: Inspur General Software Co Ltd
Priority date: 2015-08-24
Filing date: 2015-08-24
Publication date: 2015-12-30

Abstract

本发明公开了一种海量数据采集的方法，具体实现过程为：首先在子数据库安装采集客户端，该采集客户端与数据采集web服务、windows服务组成异步处理采集系统；母数据库通过该异步处理采集系统采集子数据库数据并按规则下发数据给子数据库的采集客户端。该一种海量数据采集的方法与现有技术相比，可节省沟通成本，并且将经过清洗流程采集上来的数据转换为可供分析的模式，提高了数据质量和安全性，实用性强。

Description

一种海量数据采集的方法

技术领域

本发明涉及大数据的数据采集技术领域，具体地说是一种实用性强、海量数据采集的方法。

背景技术

生产企业常常需要通过相关经销商的流向数据来了解产品销售情况，或者通过相关数据进行业务分析和开展。目前的情况是，不同生产企业在经销商那采集数据的工具种类繁多，且采集的数据格式不够统一，容易造成经销商的重复劳动，耗费交流沟通时间，并且由业务员上报的数据准确性也无法保证，可能对后续分析造成影响。

基于此，现提供一种可解决上述问题、海量数据采集的方法。使用本方法的数据采集后，可以从公司委托发起，节省沟通成本，客户端自动对经销商进销存的数据进行采集、压缩上传到集团服务器、然后再把数据根据不同的分发规则分发至集团下属各生产或销售企业中，经过清洗流程将采集上来的数据转换为可供分析的模式，提高了数据质量和安全性。

发明内容

本发明的技术任务是针对以上不足之处，提供一种实用性强、海量数据采集的方法。

一种海量数据采集的方法，其具体实现过程为：

首先在子数据库安装采集客户端，该采集客户端与数据采集web服务、windows服务组成异步处理采集系统；

母数据库通过该异步处理采集系统采集子数据库数据并按规则下发数据给子数据库的采集客户端。

所述采集客户端支持的数据库类型包括：SQLSERVER、Oracle、Mysql、Sybase。

安装有采集客户端的计算机通过数据库配置接口连接到安装母数据库的后台服务器，采集客户端对数据采集成功后，将该数据使用文件处理，这里的文件处理使用压缩文件的方式，每个压缩文件中均包含一个文本文件和一个数据描述xml文件，其中文本文件为实际采集数据，数据描述xml文件为文本文件中数据的规则；服务器将两个文件一起读取解析后再转换为可使用数据。

安装有母数据库的后台服务器通过异步排队机制支持采集客户端的并发请求操作，母数据库采集端对所有采集客户端上传的文件首先汇总到服务器的某个文件夹下，然后定期将文件更新到数据处理队列中，数据处理完毕后会将原数据包转移到另一个备份文件夹中。

上述后台服务器采用异步排队处理模式进行数据传输，且该异步处理使用以下多线程完成：

数据上传线程，检查客户端的数据文件夹中是否有已采集完毕的数据，有则加入上传队列；调用采集上传服务进行token验证，如果验证通过则将数据文件包分解成若干个24K的文件包进行传输；在传输的过程中，当遇到网络问题时，进行多次传输尝试，如果多次后仍失败，则在线程下次执行时继续尝试传输；

采集线程，检查采集任务队列中是否有待执行的采集任务；根据采集任务中的信息执行采集动作，并将采集的数据进行压缩打包，放到待上传的数据文件夹中；

自动采集时间判断线程，根据客户端设置的自动采集频率和时间，检查当前是否需要执行自动采集；如果需要执行数据采集，则在采集任务队列中增加自动采集的任务；

与服务器自动更新线程，与采集服务通信，下载最新数据采集定义、往服务器上报包括客户端版本号和经销商采集时间点的信息；

规则日志线程，线程启动时，创建全局缓存对象，构造对象时读取LOG.XML文件，将里面的dLog对象存在在_logList对象列表中；规则日志线程10秒检查一次缓存中日志对象列表有没有变化，有变化就重新生成Log.XML；

文本日志线程，线程启动时，创建全局缓存对象，线程10秒钟循环一次，声明一个StringBuild，将生成的日志信息增加到里面，写入以日期命名的Txt日志文件中；

版本自动检查线程，执行时调用采集服务，自动比对客户单文件的版本与服务端的版本；当客户端需要升级时，则调用升级程序对需要升级的文件进行下载，解压，覆盖。

在数据采集过程中，数据文件传输时使用AES加密，加密的秘钥为采集服务对每个客户端生成的一个36位GUID字符串，使得每个客户端的秘钥均不一样，保证数据传输中的安全性，具体为：

所有需要与服务器通信的客户端账号，均被分配一个唯一的密钥，每次申请通信时均需要使用唯一密钥、账号、密码验证，通过后才可进行文件传输，在数据文件传输过程中使用AES动态加密数据片段，且每次传输文件大小为24K，只有文件全部传输成功后，文件才可被使用。

后台服务器接收采集客户端的压缩文件上传，所有正在传输的文件均以.dat的后缀保存在传输目录中；当文件完成上传时，由客户端发起文件的改名操作，数据采集服务把文件后缀从.dat修改成.rar，并移动文件至文件导入工作目录中，同时通知文件导入功能。

本发明的一种海量数据采集的方法，具有以下优点：

该发明的一种海量数据采集的方法能够在企业总部采集全国下属经销商进销存数据，并按规则下发数据至下属分公司中。本系统通过已维护好的各类采集对象与方案定义，通过数据直连采集客户端与数据采集web服务、windows服务组成的整个异步处理采集系统，每天可处理千万条集团数据的采集和分发，从而节省沟通成本，提高数据质量和安全性，实用性强，适用范围广泛，易于推广。

附图说明

附图1为本发明的整体实现流程图。

附图2为本发明的采集客户端采集流程图。

附图3为本发明的后台服务器采集流程图。

具体实施方式

下面结合附图和具体实施例对本发明作进一步说明。

本发明的提供一种海量数据采集的方法，通过数据采集客户端进行海量数据采集的方式进行数据采集。

其实现过程为：

采集客户端能够支持经销商运行的复杂多样性的系统，能够支持多种数据库的类型，如SQLSERVER、Oracle、Mysql、Sybase等；。

采集客户端支持灵活部署模式，满足同一客户可向多家供货商提供数据的要求，在实际部署时支持多种那个部署模式：

全集中架构：采集客户端集中采集经销商的数据，不区分集团下属单位，进行集中抓取和数据上传。数据集中到集团后，再从集团数据库中进行数据的分发。

部分集中架构：客户端在采集经销商的数据时区分委托公司，数据进行集中抓取和上传；通过集团数据采集服务进行数据分发。

分散架构：客户端在采集经销商的数据时区分委托公司，数据进行抓取后分别上传至不同的委托公司服务器中。

数据采集服务器通过异步排队机制支持数据直连客户端的大并发请求操作，提高采集服务的可用性：采集服务端对所有客户端上传的文件首先汇总到服务器的某个文件夹下，然后定期将文件更新到数据处理队列中，数据处理完毕后会将原数据包转移到另一个备份文件夹中，减少大并发上传文件时对数据处理线程的影响，保证数据处理时的稳定性。

客户端在进行采集和文件上传时，如果发生错误，则会进行多次尝试，体现了系统高容错性支持。

数据文件在进行传输时，将文件分成多个24K大小的文件包进行传输，文件传输完毕，采集服务会将文件合并成原数据文件并使用；以24K大小传输是考虑到对两端网络的压力，对企业采集服务器CPU、IO等硬件的影响而设定的。

数据文件在传输过程中使用AES加密，加密的秘钥为采集服务对每个客户端生成的一个36位GUID字符串，使得每个客户端的秘钥均不一样，进一步保证了数据传输中的安全性。

集团模式下的数据采集支持根据系统内预制的标准规则（如：客户对应关系、产品对应关系等）自动智能下发到下级企业帐套，不需要人为干预。

更为具体的，如附图1、图2、图3所示，

安装有采集客户端的计算机通过数据库配置接口连接到安装母数据库的后台服务器，采集客户端对数据采集成功后，将该数据使用文件处理，这里的文件处理使用压缩文件的方式，每个压缩文件中均包含一个文本文件和一个数据描述xml文件，其中文本文件为实际采集数据，数据描述xml文件为文本文件中数据的规则；服务器将两个文件一起读取解析后再转换为可使用数据，压缩文件大大降低了网络带宽占用，同时极大增强系统处理能力；

按照特殊格式生成txt文件部分代码：

txtbuffer.SaveTxt(String.Format("{0}\0",

onvert.ToDateTime(drResult[k]).ToString("yyyyMMdd")));

将XML文件和txt文件生成压缩包部分代码：

publicstaticstringDataEntryName="data.txt";

publicstaticstringDataFormatEntryName="format.xml";

publicstaticintZipBufferSize=1024*24;

//使用采集客户端设置的文件参数生成压缩文件

byte[]buffer=newbyte[DCSZipConfig.ZipBufferSize];

System.Int32sizeRead=0;

while((sizeRead=StreamToZip.Read(buffer,0,buffer.Length))>0)

{

Write(buffer,0,sizeRead);

}

采集后台服务系统采用异步排队处理模式：程序使用多线程的异步处理，不同的线程处理不同的任务，如：采集数据压缩成文件、文件上传、文件导入数据库、数据分发等均可同时进行，平滑降低服务器的压力峰值；、

数据上传线程SendDataThread，检查客户端的数据文件夹中是否有已采集完毕的数据，有则加入上传队列；调用采集上传服务进行token验证，如果验证通过则将数据文件包分解成若干个24K的文件包进行传输；在传输的过程中，当遇到网络问题时，进行多次传输尝试，如果多次后仍失败，则在线程下次执行时继续尝试传输；其执行频率设置为3分钟。

采集线程CollectionThread，检查采集任务队列中是否有待执行的采集任务；根据采集任务中的信息执行采集动作，并将采集的数据进行压缩打包，放到待上传的数据文件夹中；其执行频率设置为5秒钟。

自动采集时间判断线程CollectDateTimeJudge，根据客户端设置的自动采集频率和时间，检查当前是否需要执行自动采集；如果需要执行数据采集，则在采集任务队列中增加自动采集的任务；其执行频率设置为10秒钟。

与服务器自动更新线程CollectionDownAndUploadInfo，与采集服务通信，下载最新数据采集定义、往服务器上报包括客户端版本号和经销商采集时间点的信息；其执行频率设置为12小时。

规则日志线程TableLogBuffer，线程启动时，创建全局缓存对象BufferObj，构造对象时读取LOG.XML文件，将里面的dLog对象存在在_logList对象列表中；规则日志线程10秒检查一次缓存中日志对象列表有没有变化，有变化就重新生成Log.XML；其执行频率设置为10秒钟。

文本日志线程TxtLogBuffer，线程启动时，创建全局缓存对象BufferObj，线程10秒钟循环一次，声明一个StringBuild，将生成的日志信息增加到里面，写入以日期命名的Txt日志文件中；其执行频率设置为10秒钟。

版本自动检查线程ThreadUpdate，执行时调用采集服务，自动比对客户单文件的版本与服务端的版本；当客户端需要升级时，则调用升级程序对需要升级的文件进行下载，解压，覆盖。其执行频率设置为24小时。

地方账户的安全保障：所有需要与服务器通信的客户端账号，均被分配一个唯一的密钥，每次申请通信时均需要使用唯一密钥、账号、密码验证，通过后才可进行文件传输，在数据文件传输过程中使用AES动态加密数据片段，且每次传输文件大小为24K，只有文件全部传输成功后，文件才可被使用；

传输时需要每次24K，并且进行token验证部分代码：

byte[]buffer=newbyte[24*1024];//24K

System.Int32sizeRead=0;

AESmyAES=newAES(company.CompanyGuid);

while((sizeRead=StreamToZip.Read(buffer,0,buffer.Length))>0)

{

byte[]mw=myAES.AESEncrypt(buffer,0,sizeRead);

//

company.DataDealProxy.AppendFile(strToken,buffer,buffer.Length,uploadFileTempName);

//每次传输完24K字节都要暂停1秒，这样每个客户端最大的网络带宽占用为24K字节/秒

Thread.Sleep(1000);

}

系统整体容错支持：在客户端采集过程中，如果出现错误，出错线程自身会不断尝试（限定最大尝试次数、时间间隔）；增加采集业务检查线程，定时巡检，对于巡检结果进行提示、发邮件、或自动处理，如：遇到服务器断电或网络不通等状况，当问题处理后，客户端自动会将未能按期进行采集的数据自动识别并采集；

文件传安全性及控制：输数据采集服务需要接收采集客户端的压缩文件上传，所有正在传输的文件均以.dat的后缀保存在传输目录中；一旦文件完成上传，由客户端发起文件的改名操作，数据采集服务把文件后缀从.dat修改成.rar，并移动文件至文件导入工作目录中，并同时通知文件导入功能。

系统通过下面参数来设定文件传输目录、和导入工作目录：

<addkey="TransPath"value="D:\DCS\DataTrans"/>

<addkey="DataPath"value="D:\DCS\Data\"/>

数据采集服务在宕掉时，文件传输目录中有可能会遗漏.rar后缀的待导入文件，系统会定期检查文件传输目录中的.rar后缀结束的文件，若有则会把它转移到导入工作目录中进行后续的导入处理。

在文件导入时如果发生错误，再某一个时间后会重新加入待处理队列代码：

boolinlist=_beforeDealFileList.ContainsKey(fi.Name)||_dealingFileList.ContainsKey(fi.Name);

if(!inlist)

{

//add增加对异常文件的重新处理

if(_errorFileList.ContainsKey(fi.Name))

{

//如果此文件以前导入出了错，并且上次导入时间距现在大于27分钟，就再重新加入导入列表中

if((DateTime.Now-_errorFileList[fi.Name]).TotalMinutes>27)sinfos.Add(fi.LastWriteTime.ToString("yyyy-MM-ddHH:mm:ss.fff")+fi.Name,fi.Name);

}

else

sinfos.Add(fi.LastWriteTime.ToString("yyyy-MM-ddHH:mm:ss.fff")+fi.Name,fi.Name);

}

采集方法的实施主要分为服务器端环境搭建及客户端的部署，其中客户端为绿色版，解压后可以直接使用；

上述具体实施方式仅是本发明的具体个案，本发明的专利保护范围包括但不限于上述具体实施方式，任何符合本发明的一种海量数据采集的方法的权利要求书的且任何所属技术领域的普通技术人员对其所做的适当变化或替换，皆应落入本发明的专利保护范围。

Claims

1.一种海量数据采集的方法，其特征在于，具体实现过程为：首先在子数据库安装采集客户端，该采集客户端与数据采集web服务、windows服务组成异步处理采集系统；母数据库通过该异步处理采集系统采集子数据库数据并按规则下发数据给子数据库的采集客户端。

2.根据权利要求1所述的一种海量数据采集的方法，其特征在于，所述采集客户端支持的数据库类型包括：SQLSERVER、Oracle、Mysql、Sybase。

3.根据权利要求1所述的一种海量数据采集的方法，其特征在于，安装有采集客户端的计算机通过数据库配置接口连接到安装母数据库的后台服务器，采集客户端对数据采集成功后，将该数据使用文件处理，这里的文件处理使用压缩文件的方式，每个压缩文件中均包含一个文本文件和一个数据描述xml文件，其中文本文件为实际采集数据，数据描述xml文件为文本文件中数据的规则；服务器将两个文件一起读取解析后再转换为可使用数据。

4.根据权利要求1所述的一种海量数据采集的方法，其特征在于，安装有母数据库的后台服务器通过异步排队机制支持采集客户端的并发请求操作，母数据库采集端对所有采集客户端上传的文件首先汇总到服务器的某个文件夹下，然后定期将文件更新到数据处理队列中，数据处理完毕后会将原数据包转移到另一个备份文件夹中。

5.根据权利要求4所述的一种海量数据采集的方法，其特征在于，上述后台服务器采用异步排队处理模式进行数据传输，且该异步处理使用以下多线程完成：

6.根据权利要求5所述的一种海量数据采集的方法，其特征在于，在数据采集过程中，数据文件传输时使用AES加密，加密的秘钥为采集服务对每个客户端生成的一个36位GUID字符串，使得每个客户端的秘钥均不一样，保证数据传输中的安全性，具体为：

7.根据权利要求5所述的一种海量数据采集的方法，其特征在于，后台服务器接收采集客户端的压缩文件上传，所有正在传输的文件均以.dat的后缀保存在传输目录中；当文件完成上传时，由客户端发起文件的改名操作，数据采集服务把文件后缀从.dat修改成.rar，并移动文件至文件导入工作目录中，同时通知文件导入功能。