use OffHeap memory to store temporary parquet and orc files to avoid GC

robinhood-jim · robinhood-jim · commit 383680767a81 · 2024-12-19T16:31:02.000+08:00
diff --git a/hadooptool/pom.xml b/hadooptool/pom.xml
@@ -14,18 +14,18 @@
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <aws.java.sdk.version>2.20.43</aws.java.sdk.version>
+        <flink.version>1.20.0</flink.version>
+        <cos.version>5.6.24</cos.version>
+        <bos.version></bos.version>
+        <cos.version>5.6.24</cos.version>
+        <obs.version>3.23.9</obs.version>
+        <qiniu.version>[7.16.0, 7.16.99]</qiniu.version>
+        <oss.version>3.17.0</oss.version>
+        <bce.version>0.10.353</bce.version>
+        <cassandra.version>3.11.12</cassandra.version>
+        <flink.version>1.20.0</flink.version>
     </properties>
-    <dependencyManagement>
-        <dependencies>
-            <dependency>
-                <groupId>software.amazon.awssdk</groupId>
-                <artifactId>bom</artifactId>
-                <version>${aws.java.sdk.version}</version>
-                <type>pom</type>
-                <scope>import</scope>
-            </dependency>
-        </dependencies>
-    </dependencyManagement>
+
 
     <dependencies>
         <dependency>
@@ -128,6 +128,7 @@
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>
+            <optional>true</optional>
         </dependency>
 
         <dependency>
@@ -282,17 +283,20 @@
         <dependency>
             <groupId>com.qcloud</groupId>
             <artifactId>cos_api</artifactId>
-            <version>5.6.24</version>
+            <version>${cos.version}</version>
+            <optional>true</optional>
         </dependency>
         <dependency>
             <groupId>com.aliyun.oss</groupId>
             <artifactId>aliyun-sdk-oss</artifactId>
-            <version>3.17.0</version>
+            <version>${oss.version}</version>
+            <optional>true</optional>
         </dependency>
         <dependency>
             <groupId>com.huaweicloud</groupId>
             <artifactId>esdk-obs-java-bundle</artifactId>
-            <version>3.23.9</version>
+            <version>${obs.version}</version>
+            <optional>true</optional>
             <exclusions>
                 <exclusion>
                     <groupId>org.apache.logging.log4j</groupId>
@@ -307,6 +311,8 @@
         <dependency>
             <groupId>software.amazon.awssdk</groupId>
             <artifactId>s3</artifactId>
+            <version>${aws.java.sdk.version}</version>
+            <optional>true</optional>
             <exclusions>
                 <exclusion>
                     <groupId>software.amazon.awssdk</groupId>
@@ -321,14 +327,20 @@
         <dependency>
             <groupId>software.amazon.awssdk</groupId>
             <artifactId>sso</artifactId>
+            <version>${aws.java.sdk.version}</version>
+            <optional>true</optional>
         </dependency>
         <dependency>
             <groupId>software.amazon.awssdk</groupId>
             <artifactId>ssooidc</artifactId>
+            <version>${aws.java.sdk.version}</version>
+            <optional>true</optional>
         </dependency>
         <dependency>
             <groupId>software.amazon.awssdk</groupId>
             <artifactId>apache-client</artifactId>
+            <version>${aws.java.sdk.version}</version>
+            <optional>true</optional>
             <exclusions>
                 <exclusion>
                     <groupId>commons-logging</groupId>
@@ -339,13 +351,14 @@
         <dependency>
             <groupId>com.qiniu</groupId>
             <artifactId>qiniu-java-sdk</artifactId>
-            <version>[7.16.0, 7.16.99]</version>
+            <version>${qiniu.version}</version>
             <optional>true</optional>
         </dependency>
         <dependency>
             <groupId>com.baidubce</groupId>
             <artifactId>bce-java-sdk</artifactId>
-            <version>0.10.353</version>
+            <version>${bce.version}</version>
+            <optional>true</optional>
             <exclusions>
                 <exclusion>
                     <groupId>org.eclipse.paho</groupId>
@@ -372,6 +385,12 @@
             <optional>true</optional>
         </dependency>
         <!-- s3 import -->
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-core</artifactId>
+            <version>${flink.version}</version>
+            <optional>true</optional>
+        </dependency>
 
 
     </dependencies>
diff --git a/hadooptool/src/main/java/com/robin/comm/fileaccess/fs/AbstractCloudStorageFileSystemAccessor.java b/hadooptool/src/main/java/com/robin/comm/fileaccess/fs/AbstractCloudStorageFileSystemAccessor.java
@@ -69,6 +69,13 @@ protected InputStream getInputStreamByConfig(DataCollectionMeta meta) {
     protected String getBucketName(DataCollectionMeta meta) {
         return !ObjectUtils.isEmpty(bucketName)?bucketName:meta.getResourceCfgMap().get(ResourceConst.BUCKETNAME).toString();
     }
+
+    /**
+     * Cloud storage now only support ingest InputStream ,So use ByteArrayOutputStream or use temp file to store data temporary
+     * @param meta
+     * @return
+     * @throws IOException
+     */
     protected OutputStream getOutputStream(DataCollectionMeta meta) throws IOException {
         OutputStream outputStream;
         if(!ObjectUtils.isEmpty(meta.getResourceCfgMap().get(ResourceConst.USETMPFILETAG)) && "true".equalsIgnoreCase(meta.getResourceCfgMap().get(ResourceConst.USETMPFILETAG).toString())){
diff --git a/hadooptool/src/main/java/com/robin/comm/fileaccess/iterator/ParquetFileIterator.java b/hadooptool/src/main/java/com/robin/comm/fileaccess/iterator/ParquetFileIterator.java
@@ -1,9 +1,8 @@
 package com.robin.comm.fileaccess.iterator;
 
+import com.robin.comm.fileaccess.util.ByteBufferSeekableInputStream;
 import com.robin.comm.fileaccess.util.ParquetUtil;
-import com.robin.comm.fileaccess.util.SeekableInputStream;
 import com.robin.core.base.util.Const;
-import com.robin.core.base.util.IOUtils;
 import com.robin.core.fileaccess.iterator.AbstractFileIterator;
 import com.robin.core.fileaccess.meta.DataCollectionMeta;
 import com.robin.core.fileaccess.util.AvroUtils;
@@ -13,6 +12,9 @@
 import org.apache.avro.Schema;
 import org.apache.avro.generic.GenericData;
 import org.apache.commons.io.FileUtils;
+import org.apache.commons.io.IOUtils;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.core.memory.MemorySegmentFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.parquet.ParquetReadOptions;
@@ -22,19 +24,20 @@
 import org.apache.parquet.hadoop.ParquetReader;
 import org.apache.parquet.hadoop.metadata.ParquetMetadata;
 import org.apache.parquet.hadoop.util.HadoopInputFile;
-import org.apache.parquet.hadoop.util.HadoopStreams;
 import org.apache.parquet.io.InputFile;
 import org.apache.parquet.io.LocalInputFile;
 import org.apache.parquet.schema.MessageType;
 import org.apache.parquet.schema.Type;
 import org.apache.slider.server.appmaster.management.Timestamp;
 import org.springframework.util.ObjectUtils;
 
-import java.io.ByteArrayOutputStream;
 import java.io.File;
 import java.io.IOException;
 import java.net.URI;
 import java.net.URL;
+import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
 import java.nio.file.Paths;
 import java.util.HashMap;
 import java.util.List;
@@ -50,6 +53,7 @@ public class ParquetFileIterator extends AbstractFileIterator {
     private ParquetReader<Map<String, Object>> ireader;
     private boolean useAvroEncode = false;
     private static final long maxSize = Integer.MAX_VALUE;
+    private MemorySegment segment;
 
     public ParquetFileIterator() {
         identifier = Const.FILEFORMATSTR.PARQUET.getValue();
@@ -100,9 +104,17 @@ public void beforeProcess() {
                         copyToLocal(tmpFile, instream);
                         file = new LocalInputFile(Paths.get(new URI(tmpFilePath)));
                     } else {
-                        ByteArrayOutputStream byteout = new ByteArrayOutputStream((int) size);
-                        IOUtils.copyBytes(instream, byteout, 8000);
-                        SeekableInputStream seekableInputStream = new SeekableInputStream(byteout.toByteArray());
+                        //use flink memory utils to use offHeapMemory to dump file content
+                        segment= MemorySegmentFactory.allocateOffHeapUnsafeMemory((int)size,this,new Thread(){});
+                        ByteBuffer byteBuffer=segment.getOffHeapBuffer();
+                        ReadableByteChannel channel= Channels.newChannel(instream);
+                        IOUtils.readFully(channel,byteBuffer);
+                        byteBuffer.position(0);
+                        channel.close();
+                        ByteBufferSeekableInputStream seekableInputStream=new ByteBufferSeekableInputStream(byteBuffer);
+                        /*ByteArrayOutputStream byteout = new ByteArrayOutputStream((int) size);
+                        IOUtils.copy(instream, byteout, 4096);
+                        ByteArraySeekableInputStream seekableInputStream = new ByteArraySeekableInputStream(byteout.toByteArray());*/
                         file = ParquetUtil.makeInputFile(seekableInputStream);
                     }
                 }
@@ -210,6 +222,10 @@ public void close() throws IOException {
         if (!ObjectUtils.isEmpty(preader)) {
             preader.close();
         }
+        //free offHeap memory
+        if(!ObjectUtils.isEmpty(segment)){
+            segment.free();
+        }
 
         if (!ObjectUtils.isEmpty(tmpFile)) {
             FileUtils.deleteQuietly(tmpFile);
diff --git a/hadooptool/src/main/java/com/robin/comm/fileaccess/iterator/ParquetStreamIterator.java b/hadooptool/src/main/java/com/robin/comm/fileaccess/iterator/ParquetStreamIterator.java
@@ -1,7 +1,7 @@
 package com.robin.comm.fileaccess.iterator;
 
 import com.robin.comm.fileaccess.util.ParquetUtil;
-import com.robin.comm.fileaccess.util.SeekableInputStream;
+import com.robin.comm.fileaccess.util.ByteArraySeekableInputStream;
 import com.robin.core.base.util.Const;
 import com.robin.core.base.util.IOUtils;
 import com.robin.core.fileaccess.iterator.AbstractFileIterator;
@@ -58,7 +58,7 @@ public void beforeProcess()  {
             //seek remote file to local tmp
             ByteArrayOutputStream byteout=new ByteArrayOutputStream(instream.available());
             IOUtils.copyBytes(instream,byteout,8000);
-            com.robin.comm.fileaccess.util.SeekableInputStream seekableInputStream=new SeekableInputStream(byteout.toByteArray());
+            ByteArraySeekableInputStream seekableInputStream=new ByteArraySeekableInputStream(byteout.toByteArray());
             preader = AvroParquetReader
                     .<GenericData.Record>builder(ParquetUtil.makeInputFile(seekableInputStream)).withConf(conf).build();
             fields = schema.getFields();
diff --git a/hadooptool/src/main/java/com/robin/comm/fileaccess/util/ByteArraySeekableInputStream.java b/hadooptool/src/main/java/com/robin/comm/fileaccess/util/ByteArraySeekableInputStream.java
@@ -6,14 +6,17 @@
 import java.io.ByteArrayInputStream;
 import java.io.EOFException;
 import java.io.IOException;
+import java.nio.ByteBuffer;
 
 
-public class SeekableInputStream extends ByteArrayInputStream implements Seekable, PositionedReadable {
+public class ByteArraySeekableInputStream extends ByteArrayInputStream implements Seekable, PositionedReadable {
 
-
-    public SeekableInputStream(byte[] buf) {
+    public ByteArraySeekableInputStream(byte[] buf) {
         super(buf);
     }
+    public ByteArraySeekableInputStream(ByteBuffer byteBuffer){
+        super(byteBuffer.array());
+    }
 
 
     @Override
diff --git a/hadooptool/src/main/java/com/robin/comm/fileaccess/util/ByteBufferSeekableInputStream.java b/hadooptool/src/main/java/com/robin/comm/fileaccess/util/ByteBufferSeekableInputStream.java
@@ -0,0 +1,101 @@
+package com.robin.comm.fileaccess.util;
+
+
+import org.apache.hadoop.fs.PositionedReadable;
+import org.apache.hadoop.fs.Seekable;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+
+public class ByteBufferSeekableInputStream extends InputStream implements  Seekable, PositionedReadable {
+    private ByteBuffer byteBuffer;
+    public ByteBufferSeekableInputStream(ByteBuffer byteBuffer){
+        this.byteBuffer=byteBuffer;
+    }
+    @Override
+    public int read() throws IOException {
+        if (byteBuffer.remaining() == 0) {
+            return -1;
+        }
+        return byteBuffer.get() & 0xff;
+    }
+
+    @Override
+    public int available() throws IOException {
+        return byteBuffer.remaining();
+    }
+
+    @Override
+    public int read(byte[] b, int off, int len) throws IOException {
+        if (byteBuffer.remaining() == 0) {
+            return -1;
+        }
+        if (len > byteBuffer.remaining()) {
+            len = byteBuffer.remaining();
+        }
+        byteBuffer.get(b, off, len);
+        return len;
+    }
+
+    @Override
+    public int read(byte[] b) throws IOException {
+        return read(b, 0, b.length);
+    }
+
+    @Override
+    public long skip(long n) throws IOException {
+        long newPos = byteBuffer.position() + n;
+        if (newPos > byteBuffer.remaining()) {
+            n = byteBuffer.remaining();
+        }
+        byteBuffer.position(byteBuffer.position() + (int) n);
+        return n;
+    }
+
+    @Override
+    public int read(long position, byte[] buffer, int offset, int length) throws IOException {
+        long oldPos=getPos();
+        int nread = -1;
+        try {
+            seek(position);
+            nread = read(buffer, offset, length);
+        } finally {
+            seek(oldPos);
+        }
+        return nread;
+    }
+
+    @Override
+    public void readFully(long position, byte[] buffer, int offset, int length) throws IOException {
+        int nread = 0;
+        while (nread < length) {
+            int nbytes = read(position + nread, buffer, offset + nread, length - nread);
+            if (nbytes < 0) {
+                throw new EOFException("End of file reached before reading fully.");
+            }
+            nread += nbytes;
+        }
+    }
+
+    @Override
+    public void readFully(long position, byte[] buffer) throws IOException {
+        readFully(position, buffer, 0, buffer.length);
+    }
+
+    @Override
+    public void seek(long pos) throws IOException {
+        byteBuffer.position((int)pos);
+    }
+
+    @Override
+    public long getPos() throws IOException {
+        return byteBuffer.position();
+    }
+
+    @Override
+    public boolean seekToNewSource(long l) throws IOException {
+        return false;
+    }
+}
diff --git a/hadooptool/src/main/java/com/robin/comm/fileaccess/util/MockInputStream.java b/hadooptool/src/main/java/com/robin/comm/fileaccess/util/MockInputStream.java
@@ -9,7 +9,7 @@ public class MockInputStream extends FSDataInputStream {
     MockFileSystem fs;
 
     public MockInputStream(MockFileSystem fs, byte[] streamBytes) throws IOException {
-        super(new SeekableInputStream(streamBytes));
+        super(new ByteArraySeekableInputStream(streamBytes));
         this.fs = fs;
     }
 
diff --git a/hadooptool/src/main/java/com/robin/comm/fileaccess/util/ParquetUtil.java b/hadooptool/src/main/java/com/robin/comm/fileaccess/util/ParquetUtil.java
diff --git a/hadooptool/src/main/java/com/robin/dfs/minio/MinioUtils.java b/hadooptool/src/main/java/com/robin/dfs/minio/MinioUtils.java

Original file line number	Diff line number	Diff line change
`@@ -9,7 +9,7 @@ public class MockInputStream extends FSDataInputStream {`
`9`	`9`	`MockFileSystem fs;`
`10`	`10`
`11`	`11`	`public MockInputStream(MockFileSystem fs, byte[] streamBytes) throws IOException {`
`12`		`- super(new SeekableInputStream(streamBytes));`
	`12`	`+ super(new ByteArraySeekableInputStream(streamBytes));`
`13`	`13`	`this.fs = fs;`
`14`	`14`	`}`
`15`	`15`