Load vcf bgz

src/main/java/au/csiro/variantspark/utils/FileUtils.java

@@ -0,0 +1,56 @@
+package au.csiro.variantspark.utils;
+
+import java.io.*;
+import java.util.zip.GZIPInputStream;
+import java.io.IOException;
+import htsjdk.samtools.util.BlockCompressedInputStream; 
+
+public class FileUtils {
+
+	/**
+	 * 
+	 * @param file: an input file
+	 * @return true if input file is BGZIP by check the first two byte of input file 
+	 */	
+	public static boolean isBGZFile(String filePath) {
+		/**
+		 * .vcf.bgz is type of GZP file, work well with BlockCompressedInputStream
+		 * .vcf.gz is also GZP file but get java.lang.OutOfMemoryError at java.io.InputStreamReader.read(InputStreamReader.java:184)
+		 * .vcf.bz2 is not GZP file and get java.lang.OutOfMemoryError at java.io.InputStreamReader.read(InputStreamReader.java:184)
+		 * .vcf is not GZP file and get htsjdk.samtools.SAMFormatException: at header from java.io.BufferedReader.readLine(BufferedReader.java:389)
+		*/
+	    try (BufferedInputStream bufferedInputStream = new BufferedInputStream(new FileInputStream(filePath))) {
+	        //bufferedInputStream.mark(100); // mark the current position
+	        boolean isValid = BlockCompressedInputStream.isValidFile(bufferedInputStream);
+	        //bufferedInputStream.reset(); // reset back to the marked position
+	        return isValid;
+	    } catch (IOException e) {
+	        // Handle the exception
+	        return false;
+	    }
+	}
+
+	/**
+	 * 
+	 * @param file: an input file
+	 * @return true if input file is Gzip by check the first two byte of input file 
+	 * @throws IOException
+	 */
+	public static boolean isInputGZip(final File file) throws IOException {
+		//final PushbackInputStream pb = new PushbackInputStream(input, 2);
+
+		try(final InputStream input = new FileInputStream(file)){
+			int header = input.read(); //read ID1
+	        if(header == -1)   return false;	        
+
+	        int b = input.read(); //read ID2
+	        if(b == -1)  return false;
+
+	        //ID2 * 256 + ID1 = 35615
+	        if( ( (b << 8) | header) == GZIPInputStream.GZIP_MAGIC) 
+	            return true;	         
+		}
+
+		return false;		
+	} 
+}

src/main/scala/au/csiro/variantspark/cli/args/SparkArgs.scala

@@ -2,15 +2,28 @@ package au.csiro.variantspark.cli.args
 
 import org.kohsuke.args4j.Option
 import au.csiro.pbdava.ssparkle.spark.SparkApp
+import au.csiro.variantspark.utils._
 import org.apache.spark.rdd.RDD
+import htsjdk.samtools.util.BlockCompressedInputStream
+import org.apache.hadoop.fs.Path
+import java.io.File
 
 trait SparkArgs extends SparkApp {
 
   @Option(name = "-sp", required = false, usage = "Spark parallelism (def=<default-spark-par>)",
     aliases = Array("--spark-par"))
   val sparkPar: Int = 0
 
-  def textFile(inputFile: String): RDD[String] =
-    sc.textFile(inputFile, if (sparkPar > 0) sparkPar else sc.defaultParallelism)
-
+  def textFile(inputFile: String): RDD[String] = {
+    val isBGZ = FileUtils.isBGZFile(inputFile)
+    println(inputFile + " is loading to spark RDD, isBGZFile: " + isBGZ)
+    if (isBGZ) {
+      val path = new Path(inputFile)
+      val fs = path.getFileSystem(sc.hadoopConfiguration)
+      val bgzInputStream = new BlockCompressedInputStream(fs.open(path))
+      sc.parallelize(Stream.continually(bgzInputStream.readLine()).takeWhile(_ != null).toList)
+    } else {
+      sc.textFile(inputFile, if (sparkPar > 0) sparkPar else sc.defaultParallelism)
+    }
+  }
 }