feat: parser support input stream reader

packyan · packyan · commit c033993cad6d · 2024-05-17T20:52:05.000+08:00
Signed-off-by: Deng An &lt;packyande@gmail.com&gt;
diff --git a/task-common/src/main/java/com/oppo/cloud/common/domain/job/LogPath.java b/task-common/src/main/java/com/oppo/cloud/common/domain/job/LogPath.java
@@ -19,6 +19,8 @@
 import com.oppo.cloud.common.constant.LogPathType;
 import lombok.Data;
 
+import java.io.InputStream;
+
 /**
  * LogPath Information
  */
@@ -45,6 +47,11 @@ public class LogPath {
      */
     private String logPath;
 
+    /*
+     * InputStream fot this log
+     */
+    private InputStream inputStream;
+
     public LogPath() {
 
     }
@@ -56,4 +63,10 @@ public LogPath(String protocol, String logType, LogPathType logPathType, String
         this.logPath = logPath;
     }
 
+    public LogPath(String protocol, String logType, InputStream inputStream) {
+        this.protocol = protocol;
+        this.logType = logType;
+        this.inputStream = inputStream;
+    }
+
 }
diff --git a/task-parser/src/main/java/com/oppo/cloud/parser/domain/reader/ReaderObject.java b/task-parser/src/main/java/com/oppo/cloud/parser/domain/reader/ReaderObject.java
@@ -37,7 +37,7 @@ public class ReaderObject {
     private BufferedReader bufferedReader;
     @Getter
     @Setter
-    private FSDataInputStream fsDataInputStream;
+    private InputStream inputStream;
     @Getter
     @Setter
     private FileSystem fs;
@@ -46,22 +46,22 @@ public BufferedReader getBufferedReader(String compressCodec) throws IOException
         if (bufferedReader != null) {
             return bufferedReader;
         }
-        InputStream inputStream;
+        InputStream decompressedInputStream;
         switch (compressCodec.toLowerCase(Locale.ROOT)) {
             case "lz4":
-                inputStream = new LZ4BlockInputStream(fsDataInputStream, false);
+                decompressedInputStream = new LZ4BlockInputStream(inputStream, false);
                 break;
             case "snappy":
-                inputStream = new SnappyInputStream(fsDataInputStream);
+                decompressedInputStream = new SnappyInputStream(inputStream);
                 break;
             case "zstd":
-                inputStream = new BufferedInputStream(new ZstdInputStream(fsDataInputStream), 32 * 1023);
+                decompressedInputStream = new BufferedInputStream(new ZstdInputStream(inputStream), 32 * 1023);
                 break;
             default:
-                inputStream = fsDataInputStream;
+                decompressedInputStream = inputStream;
                 break;
         }
-        bufferedReader = new BufferedReader(new InputStreamReader(inputStream));
+        bufferedReader = new BufferedReader(new InputStreamReader(decompressedInputStream));
         return bufferedReader;
     }
 
@@ -74,8 +74,8 @@ public BufferedReader getBufferedReader() throws IOException {
 
     public void close() {
         try {
-            if (fsDataInputStream != null) {
-                fsDataInputStream.close();
+            if (inputStream != null) {
+                inputStream.close();
             }
             if (bufferedReader != null) {
                 bufferedReader.close();
diff --git a/task-parser/src/main/java/com/oppo/cloud/parser/service/reader/ILogReaderFactory.java b/task-parser/src/main/java/com/oppo/cloud/parser/service/reader/ILogReaderFactory.java
@@ -27,6 +27,7 @@
 public interface ILogReaderFactory {
 
     String HDFS = "hdfs";
+    String STREAM =  "stream";
     String S3 = "s3";
 
     /**
@@ -36,6 +37,8 @@ default IReader create(LogPath logPath) throws Exception {
         switch (logPath.getProtocol()) {
             case HDFS:
                 return new HDFSReader(logPath, getNameNodeConf());
+            case STREAM:
+                return new InputStreamReader(logPath);
             default:
                 break;
         }
diff --git a/task-parser/src/main/java/com/oppo/cloud/parser/service/reader/InputStreamReader.java b/task-parser/src/main/java/com/oppo/cloud/parser/service/reader/InputStreamReader.java
@@ -0,0 +1,53 @@
+/*
+ * Copyright 2023 OPPO.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.oppo.cloud.parser.service.reader;
+
+import com.oppo.cloud.common.domain.job.LogPath;
+import com.oppo.cloud.parser.domain.reader.ReaderObject;
+
+import java.util.List;
+
+public class InputStreamReader implements IReader {
+    private final LogPath logPath;
+
+    public InputStreamReader(LogPath logPath) {
+        this.logPath = logPath;
+    }
+
+    @Override
+    public List<String> listFiles() throws Exception {
+        return null;
+    }
+
+    @Override
+    public List<String> filesPattern() throws Exception {
+        return null;
+    }
+
+    @Override
+    public ReaderObject getReaderObject() throws Exception {
+        ReaderObject readerObject = new ReaderObject();
+        readerObject.setLogPath(logPath.getLogPath());
+        readerObject.setFs(null);
+        readerObject.setInputStream(logPath.getInputStream());
+        return readerObject;
+    }
+
+    @Override
+    public List<ReaderObject> getReaderObjects() throws Exception {
+        return null;
+    }
+}
diff --git a/task-parser/src/main/java/com/oppo/cloud/parser/utils/HDFSUtil.java b/task-parser/src/main/java/com/oppo/cloud/parser/utils/HDFSUtil.java
@@ -137,7 +137,7 @@ public static ReaderObject getReaderObject(NameNodeConf nameNode, String path) t
         FSDataInputStream fsDataInputStream = fs.open(new Path(path));
         ReaderObject readerObject = new ReaderObject();
         readerObject.setLogPath(path);
-        readerObject.setFsDataInputStream(fsDataInputStream);
+        readerObject.setInputStream(fsDataInputStream);
         readerObject.setFs(fs);
         return readerObject;
     }
diff --git a/task-parser/src/main/java/com/oppo/cloud/parser/utils/JobHistoryUtil.java b/task-parser/src/main/java/com/oppo/cloud/parser/utils/JobHistoryUtil.java
@@ -20,14 +20,14 @@
 import com.oppo.cloud.parser.domain.mr.MRAppInfo;
 import com.oppo.cloud.parser.domain.reader.ReaderObject;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.hadoop.fs.FSDataInputStream;
 import org.w3c.dom.Document;
 import org.w3c.dom.Element;
 import org.w3c.dom.NodeList;
 
 import javax.xml.parsers.DocumentBuilder;
 import javax.xml.parsers.DocumentBuilderFactory;
 import java.io.ByteArrayInputStream;
+import java.io.InputStream;
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.List;
@@ -105,8 +105,8 @@ public static Map<String, String> parseJobConf(ReaderObject confReader) throws I
     public static MRAppInfo parseJobHistory(List<ReaderObject> readerObjects) throws Exception {
         JobHistoryFileInfo jobHistoryFileInfo = getJobHistoryFileInfo(readerObjects);
         Map<String, String> conMap = parseJobConf(jobHistoryFileInfo.getConfReader());
-        FSDataInputStream fsDataInputStream = jobHistoryFileInfo.getJobHistoryReader().getFsDataInputStream();
-        ReplayMREventLogs replayMREventLogs = new ReplayMREventLogs(fsDataInputStream);
+        InputStream inputStream = jobHistoryFileInfo.getJobHistoryReader().getInputStream();
+        ReplayMREventLogs replayMREventLogs = new ReplayMREventLogs(inputStream);
         replayMREventLogs.parse();
         MRAppInfo appData = replayMREventLogs.getMRAppInfo();
         appData.setConfMap(conMap);
diff --git a/task-parser/src/main/java/com/oppo/cloud/parser/utils/ReplayMREventLogs.java b/task-parser/src/main/java/com/oppo/cloud/parser/utils/ReplayMREventLogs.java
@@ -29,6 +29,7 @@
 import org.apache.avro.specific.SpecificDatumReader;
 import org.apache.hadoop.fs.FSDataInputStream;
 
+import java.io.InputStream;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -45,13 +46,14 @@ public class ReplayMREventLogs {
 
     private static final String SPECULATION = "Speculation";
 
-    public ReplayMREventLogs(FSDataInputStream in) {
-        this.in = in;
+    public ReplayMREventLogs(InputStream in) {
+        this.in = new FSDataInputStream(in);
         this.mrAppInfo = new MRAppInfo();
     }
 
 
     public void parse() throws Exception {
+
         String version = this.in.readLine();
         String eventSchema = this.in.readLine();
         Schema schema = new Schema.Parser().parse(eventSchema);
diff --git a/task-parser/src/test/java/com/oppo/cloud/parser/service/job/reader/InputStreamReaderTest.java b/task-parser/src/test/java/com/oppo/cloud/parser/service/job/reader/InputStreamReaderTest.java
@@ -0,0 +1,44 @@
+package com.oppo.cloud.parser.service.job.reader;
+
+import com.oppo.cloud.common.constant.LogType;
+import com.oppo.cloud.common.domain.job.LogPath;
+import com.oppo.cloud.common.domain.job.LogRecord;
+import com.oppo.cloud.parser.domain.job.CommonResult;
+import com.oppo.cloud.parser.domain.job.ParserParam;
+import com.oppo.cloud.parser.domain.job.SparkEventLogParserResult;
+import com.oppo.cloud.parser.service.ParamUtil;
+import com.oppo.cloud.parser.service.job.parser.IParser;
+import com.oppo.cloud.parser.service.job.parser.SimpleParserFactory;
+import com.oppo.cloud.parser.service.reader.LogReaderFactory;
+import com.oppo.cloud.parser.utils.ResourcePreparer;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.InputStream;
+import java.util.Collections;
+
+public class InputStreamReaderTest extends ResourcePreparer {
+    @Test
+    public void inputStreamReaderTest() {
+        String eventLogPath = "log/event/eventlog";
+        InputStream inputStream = this.getClass().getClassLoader().getResourceAsStream(eventLogPath);
+        SimpleParserFactory simpleParserFactory = new SimpleParserFactory();
+        LogRecord logRecord = ParamUtil.getLogRecord();
+        String logType = LogType.SPARK_EVENT.getName();
+        LogPath logPath = new LogPath();
+        logPath.setInputStream(inputStream);
+        logPath.setProtocol(LogReaderFactory.STREAM);
+        logPath.setLogType(logType);
+        logPath.setLogPath(eventLogPath);
+        ParserParam parserParam = new ParserParam(logType,
+                logRecord,
+                logRecord.getApps().get(0),
+                Collections.singletonList(logPath));
+        IParser parser = simpleParserFactory.createParserInternal(parserParam);
+        CommonResult commonResult = parser.run();
+        SparkEventLogParserResult result = (SparkEventLogParserResult) commonResult.getResult();
+        Assert.assertTrue(result.getMemoryCalculateParam().getDriverMemory() == 536870912L);
+        Assert.assertTrue(result.getMemoryCalculateParam().getAppTotalTime() == 588898L);
+
+    }
+}
diff --git a/task-parser/src/test/java/com/oppo/cloud/parser/utils/ResourcePreparer.java b/task-parser/src/test/java/com/oppo/cloud/parser/utils/ResourcePreparer.java
@@ -30,6 +30,10 @@ public class ResourcePreparer extends MiniHdfsCluster {
     private static String LOCAL_TEXT_LOG_DIR = "/log/";
     private static String HDFS_TEXT_LOG_DIR = "/log";
 
+    static {
+        ParserConfigLoader.init();
+    }
+
     @BeforeAll
     public static void prepareResources() throws IOException {
         final URL resourcesDir = ResourcePreparer.class.getResource(LOCAL_TEXT_LOG_DIR);

Original file line number	Diff line number	Diff line change
`@@ -137,7 +137,7 @@ public static ReaderObject getReaderObject(NameNodeConf nameNode, String path) t`
`137`	`137`	`FSDataInputStream fsDataInputStream = fs.open(new Path(path));`
`138`	`138`	`ReaderObject readerObject = new ReaderObject();`
`139`	`139`	`readerObject.setLogPath(path);`
`140`		`- readerObject.setFsDataInputStream(fsDataInputStream);`
	`140`	`+ readerObject.setInputStream(fsDataInputStream);`
`141`	`141`	`readerObject.setFs(fs);`
`142`	`142`	`return readerObject;`
`143`	`143`	`}`