optimize code

Author: robinhood-jim

common/src/test/java/com/robin/comm/test/TestExcelOperation.java

@@ -51,17 +51,10 @@ public void testGenerate() throws Exception {
         AbstractResIterator iterator = new AbstractResIterator() {
             Map<String, Object> map = new HashMap<>();
             int row = 0;
-
-            @Override
-            public void init() {
-
-            }
-
             @Override
             public void beforeProcess() {
 
             }
-
             @Override
             public void afterProcess() {
 

core/src/main/java/com/robin/core/compress/util/CompressDecoder.java

@@ -56,7 +56,7 @@ public static InputStream getInputStreamByCompressType(String path, InputStream
                 inputStream=new LZ4FrameInputStream(wrapInputStream(rawstream));
                 break;
             case COMPRESS_TYPE_LZMA:
-                inputStream=new LZMAInputStream(wrapInputStream(rawstream));
+                inputStream=new XZInputStream(wrapInputStream(rawstream));
                 break;
             case COMPRESS_TYPE_ZSTD:
                 inputStream=new ZstdCompressorInputStream(wrapInputStream(rawstream));

core/src/main/java/com/robin/core/compress/util/CompressEncoder.java

@@ -11,6 +11,7 @@
 import org.apache.commons.compress.compressors.zstandard.ZstdCompressorOutputStream;
 import org.tukaani.xz.LZMA2Options;
 import org.tukaani.xz.LZMAOutputStream;
+import org.tukaani.xz.XZOutputStream;
 import org.xerial.snappy.SnappyOutputStream;
 
 import java.io.BufferedOutputStream;
@@ -64,7 +65,7 @@ public static OutputStream getOutputStreamByCompressType(String path,OutputStrea
                 outputStream=new LZ4FrameOutputStream(wrapOutputStream(rawstream));
                 break;
             case COMPRESS_TYPE_LZMA:
-                outputStream=new LZMAOutputStream(wrapOutputStream(rawstream),new LZMA2Options(),false);
+                outputStream=new XZOutputStream(wrapOutputStream(rawstream),new LZMA2Options());
                 break;
             case COMPRESS_TYPE_ZSTD:
                 outputStream=new ZstdCompressorOutputStream(wrapOutputStream(rawstream));

hadooptool/src/main/java/com/robin/comm/fileaccess/iterator/ParquetFileIterator.java

@@ -72,17 +72,11 @@ public void beforeProcess() {
             if (colmeta.getResourceCfgMap().containsKey("file.useAvroEncode") && "true".equalsIgnoreCase(colmeta.getResourceCfgMap().get("file.useAvroEncode").toString())) {
                 useAvroEncode = true;
             }
+
             if (Const.FILESYSTEM.HDFS.getValue().equals(colmeta.getFsType())) {
                 conf = new HDFSUtil(colmeta).getConfig();
-                if (colmeta.getColumnList().isEmpty()) {
-                    ParquetReadOptions options = ParquetReadOptions.builder().withMetadataFilter(ParquetMetadataConverter.NO_FILTER).build();
-                    ParquetFileReader ireader = ParquetFileReader.open(HadoopInputFile.fromPath(new Path(colmeta.getPath()), conf), options);
-                    ParquetMetadata meta = ireader.getFooter();
-                    msgtype = meta.getFileMetaData().getSchema();
-                    parseSchemaByType();
-                } else {
-                    schema = AvroUtils.getSchemaFromMeta(colmeta);
-                }
+                file=HadoopInputFile.fromPath(new Path(colmeta.getPath()), conf);
+                getSchema(file,false);
                 if (!useAvroEncode) {
                     ParquetReader.Builder<Map<String, Object>> builder = ParquetReader.builder(new CustomReadSupport(), new Path(ResourceUtil.getProcessPath(colmeta.getPath()))).withConf(conf);
                     ireader = builder.build();
@@ -111,30 +105,35 @@ public void beforeProcess() {
                         file = ParquetUtil.makeInputFile(seekableInputStream);
                     }
                 }
-                if (colmeta.getColumnList().isEmpty()) {
-                    ParquetReadOptions options = ParquetReadOptions.builder().withMetadataFilter(ParquetMetadataConverter.NO_FILTER).build();
-                    ParquetFileReader ireader = ParquetFileReader.open(file, options);
-                    ParquetMetadata meta = ireader.getFooter();
-                    msgtype = meta.getFileMetaData().getSchema();
-                    parseSchemaByType();
-                    //read footer and schema,must return header
-                    file.newStream().seek(0L);
-                } else {
-                    schema = AvroUtils.getSchemaFromMeta(colmeta);
-                }
-                fields = schema.getFields();
+                getSchema(file,true);
                 if (!useAvroEncode) {
                     ireader = CustomParquetReader.builder(file, colmeta).build();
                 } else {
                     preader = AvroParquetReader.<GenericData.Record>builder(file).build();
                 }
             }
+            fields = schema.getFields();
         } catch (Exception ex) {
             logger.error("{}", ex.getMessage());
         }
 
     }
 
+    private void getSchema(InputFile file,boolean seekFrist) throws IOException {
+        if (colmeta.getColumnList().isEmpty()) {
+            ParquetReadOptions options = ParquetReadOptions.builder().withMetadataFilter(ParquetMetadataConverter.NO_FILTER).build();
+            ParquetFileReader ireader = ParquetFileReader.open(file, options);
+            ParquetMetadata meta = ireader.getFooter();
+            msgtype = meta.getFileMetaData().getSchema();
+            parseSchemaByType();
+            if(seekFrist) {
+                file.newStream().seek(0L);
+            }
+        } else {
+            schema = AvroUtils.getSchemaFromMeta(colmeta);
+        }
+    }
+
     @Override
     public boolean hasNext() {
         try {

hadooptool/src/main/java/com/robin/comm/fileaccess/writer/ProtoBufFileWriter.java

@@ -13,6 +13,7 @@
 import com.robin.core.fileaccess.writer.AbstractFileWriter;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.ObjectUtils;
+import org.tukaani.xz.LZMAOutputStream;
 
 import javax.naming.OperationNotSupportedException;
 import java.io.IOException;

pom.xml

@@ -127,7 +127,7 @@
         <svn.version>1.8.9</svn.version>
         <lzocore.version>1.0.4</lzocore.version>
         <snappy.version>1.1.2.6</snappy.version>
-        <xz.version>1.6</xz.version>
+        <xz.version>1.9</xz.version>
         <cassandradriver.version>3.6.0</cassandradriver.version>
         <graal.version>1.0.0-rc7</graal.version>
         <calcite.version>1.21.0</calcite.version>