Merge branch 'feature-2023-04' into 'developer'

Feature 2023 04

See merge request !78

gic-cloud-data-hook-service/src/main/java/com/gic/cloud/data/hook/service/HDFSUtil.java

@@ -47,7 +47,7 @@ public class HDFSUtil {
      */
     public boolean downloadFile(String srcPath,String toPath) {
         try {
-            fileSystem.copyToLocalFile(false,new Path(srcPath),new Path(toPath));
+            fileSystem.copyToLocalFile(true,new Path(srcPath),new Path(toPath));
             return true;
         } catch (IOException e) {
             logger.info("下载失败：{}",e);

gic-cloud-data-hook-service/src/main/java/com/gic/cloud/data/hook/service/impl/FlatQueryResultServiceImpl.java

@@ -105,7 +105,7 @@ public class FlatQueryResultServiceImpl implements IFlatQueryResultService {
 
 	private FlatQueryResultServiceImpl() {
 		log.debug("construct", "准备初始化 FlatQuery 查询服务");
-//		runDealHiveFile(3);
+		runDealHiveFile(3);
 //		runDistTask(3);
 		runDownloadTask(3);
 		runBalaDownloadTask(3);
@@ -864,31 +864,31 @@ public class FlatQueryResultServiceImpl implements IFlatQueryResultService {
 		//下载文件
 		String dirName="hdfs"+task.getId();
 		String path=HDFS_URL+"/"+dirName;
-		boolean downloadFlag = HDFSUtil.getInstance().downloadFile(path, SAVE_FOLDER);
-		if (!downloadFlag) {
-			logger.info("下载失败：{}-{}",path,JSONObject.toJSONString(task));
-			return;
-		}
-		List<String> xlsxFiles=new ArrayList<>();
-		AtomicInteger count=new AtomicInteger(0);
-		AtomicReference<XlsxFileInfo> currentFile=new AtomicReference<>();
-		readCsvFile(condition,dirName,(cells,titles,firstFlag)->{
-			if (count.get()==0) {
-				XlsxFileInfo xlsxFileInfo = new XlsxFileInfo();
-				xlsxFileInfo.filepath=SAVE_FOLDER + "/" + task.getId() + xlsxFiles.size() + ".xlsx";
-				currentFile.set(xlsxFileInfo);
-				xlsxFiles.add(currentFile.get().filepath);
-			}
-			saveXlsSplitNew(currentFile.get().filepath,cells,titles,currentFile.get(),count,false);
-			count.incrementAndGet();
-		});
-		//结束
-		saveXlsSplitNew(currentFile.get().filepath,null,null,currentFile.get(),count,true);
-
-		//是否压缩
-		boolean zipFlag = (xlsxFiles.size() > 1) ? true : false;
-		String cloudFileUrl=null;
 		try {
+			boolean downloadFlag = HDFSUtil.getInstance().downloadFile(path, SAVE_FOLDER);
+			if (!downloadFlag) {
+				logger.info("下载失败：{}-{}",path,JSONObject.toJSONString(task));
+				return;
+			}
+			List<String> xlsxFiles=new ArrayList<>();
+			AtomicInteger count=new AtomicInteger(0);
+			AtomicReference<XlsxFileInfo> currentFile=new AtomicReference<>();
+			readCsvFile(condition,dirName,(cells,titles,firstFlag)->{
+				if (count.get()==0) {
+					XlsxFileInfo xlsxFileInfo = new XlsxFileInfo();
+					xlsxFileInfo.filepath=SAVE_FOLDER + "/" + task.getId() + xlsxFiles.size() + ".xlsx";
+					currentFile.set(xlsxFileInfo);
+					xlsxFiles.add(currentFile.get().filepath);
+				}
+				saveXlsSplitNew(currentFile.get().filepath,cells,titles,currentFile.get(),count,false);
+				count.incrementAndGet();
+			});
+			//结束
+			saveXlsSplitNew(currentFile.get().filepath,null,null,currentFile.get(),count,true);
+
+			//是否压缩
+			boolean zipFlag = (xlsxFiles.size() > 1) ? true : false;
+			String cloudFileUrl=null;
 			String taskFileExt = task.getUseCompress().equals(Global.YES) || zipFlag ? ".zip" : task.getFormat().equals(DownloadFileFormat.CSV) ? ".csv" : ".xlsx";
 			if (zipFlag || task.getUseCompress().equals(Global.YES)) {
 				logger.info("[ runDownloadTask.run ]: {}", "准备生成自助指标压缩文件 " + task.getId() + ".zip");
@@ -908,10 +908,15 @@ public class FlatQueryResultServiceImpl implements IFlatQueryResultService {
 			task.setStatus(DownloadTaskStatus.COMPLISHED);
 			task.setOverTime(new Date());
 			task.setFilePath(cloudFileUrl);
-			DownloadTaskServiceImpl.getInstance().updateDownloadTask(task);
+
 
 		}catch (Exception e) {
 			logger.info("异常：{}",e);
+			task.setStatus(DownloadTaskStatus.ERROR);
+		}finally {
+			DownloadTaskServiceImpl.getInstance().updateDownloadTask(task);
+			//删除文件
+			delFileOrDirByTaskId(task.getId());
 		}
 	}
 
@@ -921,6 +926,35 @@ public class FlatQueryResultServiceImpl implements IFlatQueryResultService {
 	}
 
 
+	private void delFileOrDirByTaskId(String taskId){
+		File file = new File(SAVE_FOLDER);
+		File[] files = file.listFiles();
+		for (File midFile : files) {
+			if (midFile.getName().indexOf(taskId)!=-1) {
+				if (midFile.isDirectory()) {
+					delDir(midFile);
+				}else{
+					midFile.delete();
+				}
+			}
+		}
+	}
+
+	private void delDir(File dirFile) {
+		if (dirFile.getPath().startsWith(SAVE_FOLDER)) {
+			File[] files = dirFile.listFiles();
+			for (File file : files) {
+				if (file.isDirectory()) {
+					delDir(file);
+				}else{
+					file.delete();
+				}
+			}
+			dirFile.delete();
+		}
+	}
+
+
 	/**
 	 * 读物文件
 	 * @param dirName
@@ -954,8 +988,7 @@ public class FlatQueryResultServiceImpl implements IFlatQueryResultService {
 					}
 					func.deal(cells,titles,first);
 				}while ((cells = csvReader.readNext())!=null);
-
-
+				csvReader.close();
 			} catch (Exception e) {
 				logger.info("读取异常：{}",e);
 			}

gic-cloud-data-hook-service/src/test/java/Test.java

@@ -18,7 +18,7 @@ public class Test {
 
     @org.junit.Test
     public void test(){
-        DownloadTask downloadTask = downloadTaskService.getDownloadTaskById("1680778799753");
+        DownloadTask downloadTask = downloadTaskService.getDownloadTaskById("1680832767459");
         flatQueryResultService.takeFileNew(downloadTask);
         try {
             Thread.sleep(10000000000L);