feat: nearest gene lut table

src/main/resources/application.conf

@@ -7,6 +7,7 @@ input = "gs://genetics-portal-data"
 
 variant-index {
   path = ${output}"/variant-index/*"
+  nearest-genes = ${input}"/v2g/nearest_gene.tsv.gz"
 }
 
 ensembl {

src/main/scala/ot/geckopipe/Configuration.scala

@@ -16,7 +16,7 @@ case class IntervalSection(path: String)
 
 case class QTLSection(path: String)
 
-case class VariantSection(path: String)
+case class VariantSection(path: String, nearestGenes: String)
 
 case class VariantGeneSection(path: String)
 

src/main/scala/ot/geckopipe/Main.scala

@@ -60,26 +60,19 @@ class Commands(val ss: SparkSession, val sampleFactor: Double, val c: Configurat
   def dictionaries(): Unit = {
     logger.info("exec variant-gene-luts command")
 
-    val vIdx = VariantIndex.builder(c).load
-
-    logger.info("write rs_id to chr-position")
-    vIdx.selectBy(Seq("rs_id", "chr_id", "position"))
-      .orderBy(col("rs_id").asc)
-      .distinct()
-      .write
-      .option("delimiter","\t")
-      .option("header", "false")
-      .csv(c.output.stripSuffix("/").concat("/v2g-lut-rsid/"))
-
-    logger.info("write gene name to chr position")
-    val _ = EnsemblIndex(c.ensembl.geneTranscriptPairs)
-      .aggByGene
-      .orderBy(col("gene_id").asc)
-      .write
-      .option("delimiter","\t")
-      .option("header", "false")
-      .csv(c.output.stripSuffix("/").concat("/v2g-lut-gene/"))
-
+    val vIdxBuilder = VariantIndex.builder(c)
+    val vIdx = vIdxBuilder.load
+    val nearests = vIdxBuilder.loadNearestGenes.map( df => {
+      logger.info("generate variant index LUT with nearest genes (prot-cod and not prot-cod")
+      vIdx.table.join(df, VariantIndex.variantColumnNames, "left_outer")
+        .write
+        .json(c.output.stripSuffix("/").concat("/variant-index-lut/"))
+    })
+
+    nearests match {
+      case scala.util.Success(lut) => logger.info("generated variant index LUT")
+      case scala.util.Failure(ex) => logger.error(ex.getMessage)
+    }
   }
 
   def summaryStats(): Unit = {

src/main/scala/ot/geckopipe/index/VariantIndex.scala

@@ -3,9 +3,13 @@ package ot.geckopipe.index
 import com.typesafe.scalalogging.LazyLogging
 import org.apache.spark.sql._
 import org.apache.spark.sql.functions._
+import org.apache.spark.sql.types._
 import org.apache.spark.storage.StorageLevel
+import ot.geckopipe.functions.{loadFromCSV, splitVariantID}
 import ot.geckopipe.{Configuration, VEP}
 
+import scala.util.Try
+
 /** represents a cached table of variants with all variant columns
   *
   * columns as chr_id, position, ref_allele, alt_allele, variant_id, rs_id. Also
@@ -26,6 +30,13 @@ object VariantIndex {
   /** types of the columns named in variantColumnNames */
   val variantColumnTypes: List[String] = List("String", "long", "string", "string")
 
+  val nearestGenesSchema = StructType(
+    StructField("varid", StringType, false) ::
+      StructField("gene_id_prot_coding", StringType, false) ::
+      StructField("gene_id_prot_coding_distance", LongType) ::
+      StructField("gene_id", StringType, false) ::
+      StructField("gene_id_distance", LongType) :: Nil)
+
   /** this class build based on the Configuration it creates a VariantIndex */
   class Builder (val conf: Configuration, val ss: SparkSession) extends LazyLogging {
     def load: VariantIndex = {
@@ -41,6 +52,14 @@ object VariantIndex {
       }
     }
 
+    def loadNearestGenes: Try[DataFrame] = {
+      splitVariantID(loadFromCSV(conf.variantIndex.nearestGenes, nearestGenesSchema)(ss),
+        variantColName = "varid").map(df => {
+        df.drop("varid", "gene_id_prot_coding_distance", "gene_id_distance")
+          .repartitionByRange(col("chr_id").asc, col("position").asc)
+      })
+    }
+
     def build: VariantIndex = {
       logger.info("building variant index as specified in the configuration")
       val savePath = conf.variantIndex.path.stripSuffix("*")