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）
迟序之数，非出神怪，有形可检，有数可推。
—祖冲之
数学是知识的工具，亦是其他知识工具的泉源。所有研究顺序和度量的科学均和数学有关。
—笛卡儿
对于大部分非计算机专业出身的分析人员和业务人员来说，数据库领域的专业术语简直让人抓狂，非要搞得那么高深吗？大可不必。
数据科学家是数据的应用者，以最大限度来提炼数据价值为目的，不必像数据仓库开发者那样对数据的存储、结构以及数据仓库的内生技术一清二楚，但应该站在找到数据、拼接数据、使用数据的角度，大体了解数据的分布、处理逻辑，以便为分析快速地准备素材。
