大数据与机器学习:实践方法与行业案例3.4 自动加载程序的多线程实现-阿里云开发者社区

开发者社区> 华章出版社> 正文

大数据与机器学习:实践方法与行业案例3.4 自动加载程序的多线程实现

简介:

3.4 自动加载程序的多线程实现


本书中的自动加载程序使用Java+Hibernate实现,需要首先准备项目所需要的hibernate和annotation相应版本的jar包。

程序需要初始化文件状态表,即根据file_status表中的配置,一次性产生当日需要处理的文件的初始记录,这些记录会插入file_status表中,供随后的处理模块使用。

为了便于并行处理,我们将自动加载程序的整个流程划分为多个子项目:

1)扫描文件子项目,取名为ScanFiles。

2)将下载文件和解压文件归为一个Java项目,取名为DownLoadAndUnZip。

3)加载文件为另一个子项目,例如针对Oracle,可以创建一个LoadToOracle的项目,针对加载至Hive表中的场景,可以创建一个LoadToHive的项目,当然也可以放在一起用参数区分。

这些Java子项目之间通过file_status配置表进行数据通信,图3-10展示了这些子项目的信息交互。

 

图3-10 项目之间的关联关系

3.4.1 ScanFiles

ScanFiles是常驻内存进程,包含初始化文件状态和扫描文件两个功能。初始化文件状态模块在每天凌晨根据数据文件信息表(file_settings),将当日需要处理的数据文件信息初始化至数据文件状态表(file_status)中,初始化字段包含文件全名、批次日期、文件日期、文件状态(初始化为FILE_NOT_EXISTS)等,更多内容可参考前面对数据文件状态表的介绍。

扫描文件模块根据数据文件状态表中当日的初始化信息,在对应的目录中扫描数据文件,更新对应的状态信息。

代码清单3-3和代码清单3-4分别给出了初始化文件状态表的代码和扫描文件的代码。

代码清单 3-3

private static void generateTheList() {

if (fileStatusDao.getTodayStatusCnt().longValue() > 0L) {

// 设置TaskDay为today

TaskDayStr = DateCalc.formatedDate(dateFormatStr, DateCalc.today());

return;

}

FileSettingDao fileSettingDao = new FileSettingDaoImpl();

List<FileSetting> fileSettings = fileSettingDao.getAllFileSettings();

Set<Long> fileIdSet = new HashSet<Long>();

for (FileSetting fileSetting : fileSettings) {

if (fileIdSet.contains(fileSetting.getId()))

continue;

else

fileIdSet.add(fileSetting.getId());

// 文件名中的日期

String fileDate = DateCalc.formatedDate(fileSetting.getFileDateFormat(),

                 DateCalc.dateadd(DateCalc.today(),

                 fileSetting.getDateDif()));

// 当天需要下载的文件全名,不含路径

String fullName = fileSetting.getFilePrefix() + fileDate +

    fileSetting.getFileMid() + fileSetting.getFileSuffix();

FileStatus fileStatus = new FileStatus();

fileStatus.setFileSetting(fileSetting);

fileStatus.setFullName(fullName);

fileStatus.setStatus(Status.FILE_NOT_EXISTS); // 初始状态为FILE_NOT_EXISTS

fileStatus.setFileSize(0L);

fileStatus.setFileSizeCheckCnt(0);

fileStatus.setFileDate(fileDate);

fileStatus.setBatchDate(DateCalc.formatedDate(dateFormatStr, DateCalc.today()));

fileStatus.setPriority(fileSetting.getPriority());

for (TargetServer targetServer : fileSetting.getTargetServerSet()) {

fileStatus.setTargetServer(targetServer);

fileSettingDao.save(fileStatus);

}

try {

Thread.sleep(300);// 暂停0.3秒

} catch (InterruptedException e) {

logger.error(e.toString());

}

}

// 设置TaskDay为今天

TaskDayStr = DateCalc.formatedDate(dateFormatStr, DateCalc.today());

}

generateTheList()方法通过fileStatusDao.getTodayStatusCnt(). longValue() > 0L判断当天的文件状态是否已经初始化,如果已经初始化,则返回;否则通过FileSettingDao遍历所有的file_settings记录,并通过FileStatus实体对象将每条file_settings记录初始化为一个fileStatus对象,最后通过fileSettingDao.save(fileStatus)将对应的初始化文件状态记录保存至file_status表中。

generateTheList()方法仅在每天凌晨成功调用一次即可,调用成功之后,file_status表中将会插入当日需要处理的所有数据文件信息,并且文件的初始状态为FILE_NOT_EXISTS。

代码清单 3-4

public class ScanFileThread extends Thread {

private static Logger logger = Logger.getLogger(ScanFileThread.class);

private static FileStatusDao fileStatusDao = new FileStatusDaoImpl();

public void run() {

synchronized (ScanFileThread.class) {

scanFile();

}

}

private void scanFile() {

while (true) {

List<FileStatus> fileStatusList = fileStatusDao.getFileByStatus(

new Status[] { Status.FILE_NOT_EXISTS, Status.CHECKING_FTP_FILE });

for (FileStatus fileStatus : fileStatusList) {

FileSetting fileSetting = fileStatus.getFileSetting();

String remoteFileName = fileStatus.getFullName(); // 文件全名

Long fileSizeOld = fileStatus.getFileSize();

Integer fileCheckCnt = fileStatus.getFileSizeCheckCnt();

FTPClient ftpClient = FtpUtil.getFtpClient(fileSetting);

if (ftpClient == null)

continue;

try {

FTPFile[] ftpFiles = ftpClient.listFiles(remoteFileName);

if (ftpFiles.length == 0) {

ftpClient.logout();

ftpClient.disconnect();

continue;

}

if (fileStatus.getStatus().equals(

Status.FILE_NOT_EXISTS.name())) {

fileStatus.setStatus(Status.CHECKING_FTP_FILE);

} else if (fileStatus.getStatus().equals(Status.

CHECKING_FTP_FILE.name())) {

Long fileSizeNew = ftpFiles[0].getSize();

if (fileSizeNew.longValue() != fileSizeOld.

longValue()) {

fileStatus.setFileSize(fileSizeNew);

fileStatus.setFileSizeCheckCnt(0);

} else {

if (fileCheckCnt < 5)

fileStatus.setFileSizeCheckCnt(fileCheckCnt + 1);

else

fileStatus.setStatus(Status.FTP_FILE_OK);

}

}

fileStatusDao.update(fileStatus);

} catch (IOException e) {

logger.error("扫描FTP文件时出错:" + e.toString());

} finally {

if (ftpClient.isConnected()) {

try {

ftpClient.logout();

ftpClient.disconnect();

} catch (IOException e) {

logger.error("断开FTP连接时出错:" + e.

toString());

}

}

}

}

try {

Thread.sleep(15000); // sleep 15秒

} catch (InterruptedException e) {

logger.error("Thread.sleep(15000)时出错:" + e.toString());

}

  }

}

}

ScanFileThread类使用了FileStatusDao类用于数据库的读/写功能,该类的父类BaseDao提供了Hibernate的SessionFactory以及事务上的支持。

ScanFileThread类的主要方法scanFile()中用于读取文件类型为Status.FILE_NOT_EXISTS以及Status.CHECKING_FTP_FILE的全部记录,通过ftpClient.listFiles(remoteFileName)查看指定文件是否存在于FTP服务器中,一旦发现文件存在,就将文件的状态由Status.FILE_NOT_EXISTS更新为Status.CHECKING_FTP_FILE。

随后程序通过多次(每次间隔15秒)比较文件大小来确定FTP服务器上的文件是否已经上传完成,当连续5次检测到的文件大小都相同时,则认为文件已经完整上传至FTP服务器,程序将该文件的状态由Status.CHECKING_FTP_FILE更新为Status.FTP_FILE_OK,标明该文件可以开始下载了。

3.4.2 DownLoadAndUnZip

在DownLoadAndUnZip项目中,包含两个重要的Java类:GetFileThread类和UnZipThread类,它们分别负责下载文件和解压文件。

GetFileThread类的主要作用是采用多线程的方式下载文件,其核心方法是代码清单3-5中所示的ftpGetFile方法。

代码清单 3-5

private static void ftpGetFile(FileStatus fileStatus) {

FTPClient ftpClient = FtpUtil.getFtpClient(fileStatus.getFileSetting());

FileSetting fileSetting = fileStatus.getFileSetting();

String remoteFileName = fileStatus.getFullName(); // 文件全名,带路径

if (fileStatus.getStatus().equalsIgnoreCase("FTP_FILE_OK")) {

fileStatus.setStatus(Status.DOWNLOAD_START); // 更新下载状态

fileStatusDao.update(fileStatus);

String localBaseDir = globalConfig.get("localBaseDir").getConfigContent();

String tableName = fileSetting.getLoadConfig().getTableName();

String ftpPath = fileSetting.getFtpPath();

Date fileDate = DateCalc.dateadd(DateCalc.today(), fileSetting.getDateDif());

String fileDateStr = DateCalc.formatedDate(fileSetting.

getFileDateFormat(), fileDate);

String localDir = null; // 文件下载后,存放的本地目录

if (ftpPath.equals(""))

localDir = localBaseDir + "/" + tableName + "/" + fileDateStr;

else

localDir = localBaseDir + "/" + tableName + "/" + ftpPath +

"/" + fileDateStr;

// 创建指定路径

File ldir = new File(localDir);

if (!ldir.exists())

ldir.mkdirs();

FileOutputStream fos = null;

try {

fos = new FileOutputStream(ldir.getAbsolutePath() + "/" +

remoteFileName);

} catch (FileNotFoundException e) {

logger.error("FileOutputStream:" + e.toString());

}

try {

ftpClient.setFileType(FTPClient.BINARY_FILE_TYPE);

ftpClient.setBufferSize(1024000);

String downLoadFilePath = null;

if (ftpClient.retrieveFile(remoteFileName, fos)) {// 下载完成

downLoadFilePath = localDir + "/" + fileStatus.getFullName();

fos.close();

fileStatus.setFilePath(downLoadFilePath);

fileStatus.setStatus(Status.DOWNLOAD_FINISHED);

fileStatusDao.update(fileStatus);

} else {

fileStatus.setStatus(Status.DOWNLOAD_FAILED); // 更新下载状态

fileStatusDao.update(fileStatus);

}

} catch (IOException e) {

fileStatus.setStatus(Status.DOWNLOAD_FAILED); // 更新下载状态

fileStatus.setLogInfo(e.toString());

fileStatusDao.update(fileStatus);

logger.error("下载文件出错:" + e.toString());

} finally {

if (ftpClient != null)

try {

ftpClient.logout();

ftpClient.disconnect();

} catch (IOException e) {

logger.error("ftpClient.disconnect()出错:" + e.

toString());

}

}

}

}

该方法用于检测状态为FTP_FILE_OK的数据文件,根据配置信息自动创建以文件对应表名、FTP远程路径和文件日期拼接的本地文件存放路径(localDir = localBaseDir + "/" + tableName + "/" + ftpPath + "/" + fileDateStr;),然后将文件下载至上述创建的目录中。在下载过程中,同时更新文件的下载状态,并记录文件下载完成后的保存路径。

UnZipThread类的作用是将文件状态为DOWNLOAD_FINISHED且需要解压缩的数据文件(通过文件后缀区分)进行解压,并记录解压后的文件名称和路径。代码清单3-6展示了UnZipThread类的主要方法unzip()。

代码清单 3-6

public static void unzip(FileStatus fileStatus) {

if (!fileStatus.getStatus().equals("DOWNLOAD_FINISHED")){

logger.error("File Status not yet in DOWNLOAD_FINISHED!");

return;

}

FileSetting fileSetting = fileStatus.getFileSetting();

String fileSuffix = fileSetting.getFileSuffix().toLowerCase();

fileStatus.setStatus(Status.UNZIP_START);

fileStatus.setUpdateTime(new Timestamp(new Date().getTime()));

fileStatusDao.update(fileStatus);

if (fileSuffix.equals(".zip"))

zip(fileStatus);

else if (fileSuffix.equals(".gz"))

gz(fileStatus);

}

unzip()方法支持.zip和.gz两种压缩方式,如果压缩文件是其他压缩方式,比如.gzip,那么需要自行编写针对.gzip格式文件的解压代码。

unzip()方法调用成功后,会根据实际解压结果更新文件状态为UNZIP_FAILED或者UNZIP_FINISHED,如果解压成功,则记录解压后的文件名称及文件路径。

3.4.3 LoadToHive

LoadToHive用于将load_config中db_type='hive'的数据文件导入对应的Hive表中,并且根据load_type字段进行不同的加载处理。代码清单3-7展示了LoadToHive的核心代码。

代码清单 3-7

private void loadToTable(TargetServer targetServer,FileStatus fileStatus) {

fileStatus.setStatus(Status.LOAD_START);

fileStatusDao.update(fileStatus);

String loadType = fileStatus.getFileSetting().

getLoadConfig().getLoadType();

String hdfsBaseDir = targetServer.getHdfsBaseDir();

String hdfsDir = null;

if (loadType.equalsIgnoreCase("added"))

hdfsDir = hdfsBaseDir + "/" +

ileStatus.getFileSetting().getLoadConfig().getTableName() + "/" +

fileStatus.getFileDate();

else if (loadType.equalsIgnoreCase("covered"))

hdfsDir = hdfsBaseDir + "/" +

fileStatus.getFileSetting().getLoadConfig().getTableName();

if (!AddToHive.createHdfsDir(fileStatus, hdfsDir)) //创建hdfs目录

return;

if (loadType.equalsIgnoreCase("added")) { // 增量,将新的partition加入Hive表中

if (!AddToHive.addPartitionToHive(fileStatus, hdfsDir))

return;

} else if (loadType.equalsIgnoreCase("covered")) { // 全量,将Hive表中的历史数据删除

if (!AddToHive.delOldDataOfHive(fileStatus, hdfsDir))

return;

}

AddToHive.copyDataToHdfs(fileStatus, hdfsDir);// 将Linux本地文件复制到hdfs目录

}

LoadToHive首先创建hdfs目录,然后判断加载方式,如果加载方式是“added”(增量),则在hive表中增加一个partition(该partition指向之前创建的hdfs目录),再将数据文件复制到hdfs目录,完成Hive表增量更新的过程;如果是“covered”(全量)加载方式,则先将历史数据删除,然后将新的数据文件复制到hdfs目录即可。

3.4.4 LoadToOracle

LoadToOracle是自动加载程序在Oracle数据库上的实现。与Hive数据仓库的不同之处在于,Oracle数据库(以及其他关系型数据库)有其自身的DBMS体系,因此其数据加载方式一般采用其自身提供的实用工具,LoadToOracle调用Oracle的sqlldr进行数据批量加载(sqlldr请参考第2章关于Oracel:sqlldr的内容)。

LoadToOracle与LoadToHive的另一个主要不同点在于,数据文件在加载到目标数据库表中之前,会先创建一个中间表(在原表名后加上temp),数据文件先加载至temp表中,然后再分别根据增量或者全量的加载方式分别处理(参考1.2.2节中关于数据更新规则的内容)。

代码清单3-8给出了LoadToOracle的核心代码。

代码清单 3-8

private void loadToTable(TargetServer targetServer, FileStatus fileStatus) {

fileStatus.setStatus(Status.LOAD_START);

fileStatusDao.update(fileStatus);

String loadType = fileStatus.getFileSetting().getLoadConfig().getLoadType();

String tableName = fileStatus.getFileSetting().getLoadConfig().getTableName();

String createTempTableRes = OracleUtil.createTempTable(tableName);// 创建temp表

if (!createTempTableRes.equals("")) {

fileStatus.setStatus(Status.LOAD_FAILED);

fileStatus.setLogInfo(createTempTableRes);

fileStatusDao.update(fileStatus);

return;

}

if (!loadIntoTempTable(fileStatus)) { // 加载数据至temp表

fileStatus.setStatus(Status.LOAD_FAILED);

fileStatusDao.update(fileStatus);

return;

}

if (loadType.equalsIgnoreCase("added")) { // 增量更新

// temp表中数据insert到原表,并将temp表drop

if (OracleUtil.insertAndDropTable(tableName))

fileStatus.setStatus(Status.LOAD_FINISHED);

else

fileStatus.setStatus(Status.LOAD_FAILED);

fileStatusDao.update(fileStatus);

} else if (loadType.equalsIgnoreCase("covered")) { // 全量更新

OracleUtil.dropAndRenameTable(tableName); //drop原表并将temp表重命名为原表

}

}

当数据文件较大时,批量加载耗时较长,体现在代码上,loadIntoTempTable(fileStatus)将花费大量时间等待sqlldr批量加载命令返回(可能达1个小时以上,取决于数据文件的大小以及Oracle数据库服务器的硬件资源),这个过程中,file_status表中的文件状态一直不发生改变,这会让人有些担心。

一个改进的办法是,在等待sqlldr批量加载命令返回的过程中,通过count对应temp表中数据的条数来反馈加载进程,可以每隔5分钟count一次temp表,并将对应的行数记录下来,这样就可以直观感受到加载程序是否正在正常工作。

3.4.5 自动加载程序的部署架构

至此,我们已经完成了自动加载程序的设计和实现过程,现在看一下自动加载程序的部署框架。

将自动加载程序的主要组成项目ScanFiles、DownLoadAndUnZip和加载(LoadToOracle、LoadToHive)导出为可执行jar包,得到ScanFiles.jar、DownLoadAndUnZip.jar、LoadToOracle. jar、LoadToHive.jar。这些jar包各自专注于完成自己的功能,要完成自动加载程序的整个过程,需要这些可执行jar包之间相互协调工作,图3-11展示了这些jar包的部署架构。

 

图3-11 自动加载程序的部署架构图

如图3-11所示,将可执行jar文件复制到相应的服务器上,然后通过java-jar命令启动这些可执行jar文件即完成部署。例如在Linux服务器上,通过执行命令“java-jar/home/queziyang/jars/ScanFiles-1.0.jar &”,即在后台启动ScanFiles进程,通过jps命令可以查看对应的进程信息。

数据缓冲区上需要部署ScanFiles.jar,所有后续进程均依赖于ScanFiles.jar的产出。

每个目标服务器(或服务器集群)上都需要部署DownLoadAndUnZip.jar,它将对应的数据文件从FTP服务器上下载至目标服务器,并在目标服务器上完成文件解压。注意到,文件的解压缩并没有在FTP服务器上进行,而是下载至本地后解压,这样做是很有必要的,因为文件解压需要占用大量CPU和I/O资源,因此在FTP服务器上执行这项任务会严重影响数据缓冲区的正常功能。

3.4.6 程序的维护和优化

数据文件在目标服务器上解压后,自动加载程序并未将文件删除。为了节省服务器的存储空间,定期删除本地服务器上保留的数据文件是非常必要的。可以通过批处理的方式定期删除历史数据文件,也可以在LoadToOracle或者LoadToHive中加入删除数据文件的功能,即当数据文件成功加载后,将原始数据文件以及解压后的数据文件删除。

另一个优化点是可以将上述配置数据库的过程可视化,即在上述Java项目的基础上包装成Java Web项目,这样第一部分

数据与平台篇

(Data & Infrastructures)

迟序之数,非出神怪,有形可检,有数可推。

—祖冲之

数学是知识的工具,亦是其他知识工具的泉源。所有研究顺序和度量的科学均和数学有关。

—笛卡儿

 

对于大部分非计算机专业出身的分析人员和业务人员来说,数据库领域的专业术语简直让人抓狂,非要搞得那么高深吗?大可不必。

数据科学家是数据的应用者,以最大限度来提炼数据价值为目的,不必像数据仓库开发者那样对数据的存储、结构以及数据仓库的内生技术一清二楚,但应该站在找到数据、拼接数据、使用数据的角度,大体了解数据的分布、处理逻辑,以便为分析快速地准备素材。

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

分享:

华章出版社

官方博客
官网链接