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

Fix 2023 04

See merge request !98

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

@@ -16,6 +16,8 @@ public class HDFSUtil {
     private static HDFSUtil hdfsUtil=null;
     private static FileSystem fileSystem=null;
 
+    public static final String HDFS_URL = "/data/hook";
+
     public static HDFSUtil getInstance(){
         if (hdfsUtil==null) {
             synchronized (HDFSUtil.class) {
@@ -47,11 +49,35 @@ public class HDFSUtil {
      */
     public boolean downloadFile(String srcPath,String toPath) {
         try {
-            fileSystem.copyToLocalFile(true,new Path(srcPath),new Path(toPath));
+            Config appConfig = ConfigService.getAppConfig();
+            Integer delFlag = appConfig.getIntProperty("del.hive.flag", 1);
+            fileSystem.copyToLocalFile(delFlag.intValue()==1?true:false,new Path(srcPath),new Path(toPath));
             return true;
         } catch (IOException e) {
             logger.info("下载失败：{}",e);
             return false;
         }
     }
+
+    public boolean deleteFile(String srcPath){
+        try {
+            fileSystem.delete(new Path(srcPath),true);
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+        return true;
+    }
+
+    public String getHdfsName(String id){
+        //下载文件
+        String dirName="hdfs"+id;
+        return dirName;
+    }
+
+    public String getHdfsPath(String id){
+        //下载文件
+        String dirName=getHdfsName(id);
+        String path=HDFS_URL+"/"+dirName;
+        return path;
+    }
 }

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

@@ -10,6 +10,7 @@ import com.gic.cloud.data.hook.api.entity.DownloadTaskStatus;
 import com.gic.cloud.data.hook.api.entity.FreeQueryTaskCondition;
 import com.gic.cloud.data.hook.api.entity.Global;
 import com.gic.cloud.data.hook.api.service.IDownloadTaskService;
+import com.gic.cloud.data.hook.service.HDFSUtil;
 import com.gic.cloud.data.hook.service.MysqlHelper;
 import com.gic.cloud.data.hook.service.dao.DownloadRecordDao;
 import com.gic.cloud.data.hook.service.dao.DownloadTaskDao;
@@ -165,6 +166,10 @@ public class DownloadTaskServiceImpl implements IDownloadTaskService {
 			}
 			this.downloadTaskDao.deleteDownloadTask(task.getId());
 			CloudFileUtil.delFileByUrl(task.getFilePath());
+
+			String hdfsPath = HDFSUtil.getInstance().getHdfsPath(task.getId());
+			HDFSUtil.getInstance().deleteFile(hdfsPath);
+
 			return true;
 		} else return false;
 	}

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

@@ -78,7 +78,7 @@ public class FlatQueryResultServiceImpl implements IFlatQueryResultService {
 	public static final String SAVE_FOLDER = "/usr/local/data-hook-file";
 //	public static final String SAVE_FOLDER = "D:\\testorder";
 
-	public static final String HDFS_URL = "/data/hook";
+//	public static final String HDFS_URL = "/data/hook";
 
 	public static final String LOCK_KEY="data:hook:hive";
 
@@ -930,8 +930,10 @@ public class FlatQueryResultServiceImpl implements IFlatQueryResultService {
 		task.setDownloadWay(-1);
 		task.setFieldSize(condition.getAllFields().size());
 		//下载文件
-		String dirName="hdfs"+task.getId();
-		String path=HDFS_URL+"/"+dirName;
+		String dirName=HDFSUtil.getInstance().getHdfsName(task.getId());
+		String path=HDFSUtil.getInstance().getHdfsPath(task.getId());
+
+		HDFSUtil.getInstance().getHdfsName(task.getId());
 		try {
 			StopWatch stopWatch = StopWatch.create("down");
 			stopWatch.start();
@@ -945,9 +947,10 @@ public class FlatQueryResultServiceImpl implements IFlatQueryResultService {
 			logger.info("下载耗时：{}",stopWatch.getLastTaskTimeMillis());
 			stopWatch.start();
 			List<String> xlsxFiles=new ArrayList<>();
+			AtomicInteger totalCount=new AtomicInteger(0);
 			AtomicInteger count=new AtomicInteger(0);
 			AtomicReference<XlsxFileInfo> currentFile=new AtomicReference<>();
-			readCsvFile(condition,dirName,(cells,titles,firstFlag)->{
+			readJsonFile(condition,dirName,(cells,titles,firstFlag)->{
 				if (count.get()==0) {
 					XlsxFileInfo xlsxFileInfo = new XlsxFileInfo();
 					xlsxFileInfo.filepath=SAVE_FOLDER + "/" + task.getId() + xlsxFiles.size() + ".xlsx";
@@ -957,9 +960,12 @@ public class FlatQueryResultServiceImpl implements IFlatQueryResultService {
 				saveXlsSplitNew(currentFile.get().filepath,cells,titles,currentFile.get(),count,false,queryDataType);
 			});
 			//结束
+			Integer limitSize = FileUtil.getLimitSize();
+			Integer total= (xlsxFiles.size()-1)*limitSize+count.get();
+
 			saveXlsSplitNew(currentFile.get().filepath,null,null,currentFile.get(),count,true,queryDataType);
 			stopWatch.stop();
-			logger.info("写入本地excel耗时：{}",stopWatch.getLastTaskTimeMillis());
+			logger.info("写入本地excel耗时：{}, 数量: {}-》{}",stopWatch.getLastTaskTimeMillis(),task.getAmount(),total);
 			stopWatch.start();
 			//是否压缩
 			boolean zipFlag = (xlsxFiles.size() > 1) ? true : false;
@@ -986,6 +992,7 @@ public class FlatQueryResultServiceImpl implements IFlatQueryResultService {
 			task.setFilePath(cloudFileUrl);
 
 
+
 		}catch (Exception e) {
 			logger.info("异常：{}",e);
 			task.setStatus(DownloadTaskStatus.ERROR);
@@ -1071,6 +1078,59 @@ public class FlatQueryResultServiceImpl implements IFlatQueryResultService {
 				csvReader.close();
 			} catch (Exception e) {
 				logger.info("读取异常：{}",e);
+				throw new RuntimeException(e);
+			}
+		}
+	}
+
+
+	/**
+	 * 读物文件
+	 * @param dirName
+	 * @param func
+	 */
+	private void readJsonFile(FlatQueryTaskCondition condition, String dirName, DownloadFunc func){
+		File file = new File(SAVE_FOLDER+"/"+dirName);
+		File[] files = file.listFiles();
+		List<FlatQueryCondition> titles=null;
+		List<FlatQueryCondition> conditions = condition.getConditions();
+		List<String> keys = conditions.stream().map(mid -> mid.getFieldMark()).collect(Collectors.toList());
+		Map<String, FlatQueryCondition> columnInfoMap = conditions.stream().collect(Collectors.toMap(mid -> mid.getFieldMark(), mid ->mid));
+		List<File> fileList = Arrays.stream(files).sorted(Comparator.comparing(File::getName)).collect(Collectors.toList());
+		for (File midFile : fileList) {
+			if (!midFile.getName().endsWith("json")) {
+				continue;
+			}
+			try {
+				BufferedReader reader=new BufferedReader(new FileReader(midFile));
+				boolean first=true;
+				Exception exception=null;
+				try {
+					String line = reader.readLine();
+					while (line != null) {
+						List<String> cellList=new ArrayList<>();
+						JSONObject jsonObject = JSONObject.parseObject(line);
+						for (String key : keys) {
+							String cellVal = jsonObject.getString(key);
+							cellList.add(cellVal);
+						}
+						String[] cells = cellList.toArray(new String[]{});
+						func.deal(cells,conditions,first);
+						first=false;
+						line = reader.readLine();
+					}
+					logger.info("读取结束：{}",midFile.getName());
+				} catch (Exception e) {
+					exception=e;
+				}finally {
+					reader.close();
+				}
+				if (exception!=null) {
+					throw exception;
+				}
+			} catch (Exception e) {
+				logger.info("读取异常：{}",e);
+				throw new RuntimeException(e);
 			}
 		}
 	}