Set tmp dir for compression action by scheduler and make bufSize configurable with human-readable value (#2070)

Author: rui-mo

smart-hadoop-support/smart-hadoop/src/main/java/org/smartdata/hdfs/action/CompressionAction.java

@@ -33,6 +33,7 @@
 import org.smartdata.conf.SmartConfKeys;
 import org.smartdata.model.CompressionFileInfo;
 import org.smartdata.model.CompressionFileState;
+import org.smartdata.utils.StringUtil;
 
 import java.io.IOException;
 import java.io.InputStream;
@@ -63,7 +64,6 @@ public class CompressionAction extends HdfsAction {
   public static final String BUF_SIZE = "-bufSize";
   public static final String COMPRESS_IMPL = "-compressImpl";
   private static List<String> compressionImplList = Arrays.asList("Lz4","Bzip2","Zlib","snappy");
-  private static final String COMPRESS_DIR = "/system/ssm/compress_tmp";
 
   private String filePath;
   private Configuration conf;
@@ -78,6 +78,9 @@ public class CompressionAction extends HdfsAction {
   private CompressionFileInfo compressionFileInfo;
   private CompressionFileState compressionFileState;
 
+  private String compressionTmpPath;
+  public static final String COMPRESSION_TMP = "-compressionTmp";
+
   @Override
   public void init(Map<String, String> args) {
     super.init(args);
@@ -89,14 +92,17 @@ public void init(Map<String, String> args) {
         SmartConfKeys.SMART_COMPRESSION_MAX_SPLIT,
         SmartConfKeys.SMART_COMPRESSION_MAX_SPLIT_DEFAULT);
     this.xAttrName = SmartConstants.SMART_FILE_STATE_XATTR_NAME;
-
     this.filePath = args.get(FILE_PATH);
-    if (args.containsKey(BUF_SIZE)) {
-      this.UserDefinedbuffersize = Integer.valueOf(args.get(BUF_SIZE));
+    if (args.containsKey(BUF_SIZE) && !args.get(BUF_SIZE).isEmpty()) {
+      this.UserDefinedbuffersize = (int) StringUtil.parseToByte(args.get(BUF_SIZE));
     }
     if (args.containsKey(COMPRESS_IMPL)) {
       this.compressionImpl = args.get(COMPRESS_IMPL);
     }
+    if (args.containsKey(COMPRESSION_TMP)) {
+      // this is a temp file kept for compressing a file.
+      this.compressionTmpPath = args.get(COMPRESSION_TMP);
+    }
   }
 
   @Override
@@ -120,8 +126,6 @@ protected void execute() throws Exception {
     if (srcFile.getLen() == 0) {
       compressionFileInfo = new CompressionFileInfo(false, compressionFileState);
     } else {
-      String tempPath = COMPRESS_DIR + filePath + "_" + "aid" + getActionId() +
-          "_" + System.currentTimeMillis();
       short replication = srcFile.getReplication();
       long blockSize = srcFile.getBlockSize();
       long fileSize = srcFile.getLen();
@@ -142,12 +146,12 @@ protected void execute() throws Exception {
 
       DFSInputStream dfsInputStream = dfsClient.open(filePath);
 
-      OutputStream compressedOutputStream = dfsClient.create(tempPath,
+      OutputStream compressedOutputStream = dfsClient.create(compressionTmpPath,
           true, replication, blockSize);
       compress(dfsInputStream, compressedOutputStream);
-      HdfsFileStatus destFile = dfsClient.getFileInfo(tempPath);
+      HdfsFileStatus destFile = dfsClient.getFileInfo(compressionTmpPath);
       compressionFileState.setCompressedLength(destFile.getLen());
-      compressionFileInfo = new CompressionFileInfo(true, tempPath, compressionFileState);
+      compressionFileInfo = new CompressionFileInfo(true, compressionTmpPath, compressionFileState);
     }
     compressionFileState.setBufferSize(bufferSize);
     appendLog("Final compression bufferSize = " + bufferSize);

smart-hadoop-support/smart-hadoop/src/main/java/org/smartdata/hdfs/scheduler/CompressionScheduler.java

@@ -24,6 +24,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.smartdata.SmartContext;
+import org.smartdata.conf.SmartConf;
+import org.smartdata.conf.SmartConfKeys;
 import org.smartdata.hdfs.HadoopUtil;
 import org.smartdata.hdfs.action.HdfsAction;
 import org.smartdata.metastore.MetaStore;
@@ -34,6 +36,8 @@
 import org.smartdata.model.FileState;
 import org.smartdata.model.CompressionFileState;
 import org.smartdata.model.LaunchAction;
+import org.smartdata.model.action.ScheduleResult;
+import org.smartdata.protocol.message.LaunchCmdlet;
 
 import java.io.IOException;
 import java.net.URI;
@@ -48,15 +52,25 @@ public class CompressionScheduler extends ActionSchedulerService {
   private final URI nnUri;
   private MetaStore metaStore;
   private static final List<String> actions = Arrays.asList("compress");
+  public static String COMPRESSION_DIR;
+  public static final String COMPRESSION_TMP = "-compressionTmp";
+  public static final String COMPRESSION_TMP_DIR = "compress_tmp/";
+  private SmartConf conf;
 
   public static final Logger LOG =
       LoggerFactory.getLogger(CompressionScheduler.class);
 
   public CompressionScheduler(SmartContext context, MetaStore metaStore)
       throws IOException {
     super(context, metaStore);
+    this.conf = context.getConf();
     this.metaStore = metaStore;
     nnUri = HadoopUtil.getNameNodeUri(getContext().getConf());
+
+    String ssmTmpDir = conf.get(
+        SmartConfKeys.SMART_WORK_DIR_KEY, SmartConfKeys.SMART_WORK_DIR_DEFAULT);
+    ssmTmpDir = ssmTmpDir + (ssmTmpDir.endsWith("/") ? "" : "/");
+    CompressionScheduler.COMPRESSION_DIR = ssmTmpDir + COMPRESSION_TMP_DIR;
   }
 
   @Override
@@ -136,6 +150,16 @@ public boolean onSubmit(CmdletInfo cmdletInfo, ActionInfo actionInfo, int action
     }
   }
 
+  @Override
+  public ScheduleResult onSchedule(CmdletInfo cmdletInfo, ActionInfo actionInfo,
+                                   LaunchCmdlet cmdlet, LaunchAction action, int actionIndex) {
+    // For compression, add compressionTmp argument
+    String tmpName = createTmpName(action);
+    action.getArgs().put(COMPRESSION_TMP, COMPRESSION_DIR + tmpName);
+    actionInfo.getArgs().put(COMPRESSION_TMP, COMPRESSION_DIR + tmpName);
+    return ScheduleResult.SUCCESS;
+  }
+
   @Override
   public void onActionFinished(CmdletInfo cmdletInfo, ActionInfo actionInfo, int actionIndex) {
     if (actionInfo.isFinished()) {

smart-hadoop-support/smart-hadoop/src/test/java/org/smartdata/hdfs/action/TestCompressionAction.java

@@ -40,13 +40,18 @@ public void init() throws Exception {
     super.init();
   }
 
-  protected void compressoin(String filePath, long bufferSize) throws IOException {
+  protected void compression(String filePath, String bufferSize) throws IOException {
     CompressionAction compressionAction = new CompressionAction();
     compressionAction.setDfsClient(dfsClient);
     compressionAction.setContext(smartContext);
     Map<String, String> args = new HashMap<>();
     args.put(compressionAction.FILE_PATH, filePath);
-    args.put(compressionAction.BUF_SIZE, "" + bufferSize);
+    args.put(compressionAction.BUF_SIZE, bufferSize);
+    // set a tmp dir for compression
+    String COMPRESS_DIR = "/system/ssm/compress_tmp";
+    String tempPath = COMPRESS_DIR + filePath + "_" + "aid" + compressionAction.getActionId()
+        + "_" + System.currentTimeMillis();
+    args.put(compressionAction.COMPRESSION_TMP, tempPath);
 //    args.put(CompressionAction.COMPRESS_IMPL, "Lz4");
 //    args.put(CompressionAction.COMPRESS_IMPL,"Bzip2");
 //    args.put(CompressionAction.COMPRESS_IMPL,"Zlib");
@@ -70,24 +75,24 @@ public void testInit() throws IOException {
 
   @Test
   public void testExecute() throws Exception {
-
     String filePath = "/testCompressFile/fadsfa/213";
     int bufferSize = 1024 * 128;
 //    String compressionImpl = "Lz4";
 //    String compressionImpl = "Bzip2";
 //    String compressionImpl = "Zlib";
     byte[] bytes = TestCompressionAction.BytesGenerator.get(bufferSize);
-
     short replication = 3;
     long blockSize = DEFAULT_BLOCK_SIZE;
+
     // Create HDFS file
     OutputStream outputStream = dfsClient.create(filePath, true,
       replication, blockSize);
     outputStream.write(bytes);
     outputStream.close();
 
     // Generate compressed file
-    compressoin(filePath, bufferSize);
+    String bufferSizeForCompression = "10MB";
+    compression(filePath, bufferSizeForCompression);
 
     // Check HdfsFileStatus
     HdfsFileStatus fileStatus = dfsClient.getFileInfo(filePath);