05_glow_vcf_deltatable_germline.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import pyspark\n",
    "import glow\n",
    "from delta import *\n",
    "from pyspark.sql.functions import explode, col, size"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "builder = pyspark.sql.SparkSession.builder.appName(\"GlowDeltalakeETL\") \\\n",
    "    .config(\"spark.hadoop.io.compression.codecs\", \"io.projectglow.sql.util.BGZFCodec\") \\\n",
    "    .config(\"spark.sql.extensions\", \"io.delta.sql.DeltaSparkSessionExtension\") \\\n",
    "    .config(\"spark.sql.catalog.spark_catalog\", \"org.apache.spark.sql.delta.catalog.DeltaCatalog\")\n",
    "\n",
    "extra_packages = [\n",
    "    \"io.projectglow:glow-spark3_2.12:1.2.1\",\n",
    "]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = configure_spark_with_delta_pip(builder, extra_packages=extra_packages).getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = glow.register(spark)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "# Page Break"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# ETL Germline VCF and Save as Deltatable\n",
    "\n",
    "* NOTE: This is repetition of previous Somatic step."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "bcbio_germline_src = \"./data/bcbio_giab_somatic/na12878-na24385-germline-hg38-truth.vcf.gz\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "bcbio_germline_df = spark.read.format(\"vcf\").load(bcbio_germline_src)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- contigName: string (nullable = true)\n",
      " |-- start: long (nullable = true)\n",
      " |-- end: long (nullable = true)\n",
      " |-- names: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- referenceAllele: string (nullable = true)\n",
      " |-- alternateAlleles: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- qual: double (nullable = true)\n",
      " |-- filters: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- splitFromMultiAllelic: boolean (nullable = true)\n",
      " |-- INFO_platformnames: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- INFO_callsetwithotheruniqgenopassing: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- INFO_callsetnames: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- INFO_AC: array (nullable = true)\n",
      " |    |-- element: integer (containsNull = true)\n",
      " |-- INFO_varType: string (nullable = true)\n",
      " |-- INFO_DPSum: integer (nullable = true)\n",
      " |-- INFO_datasetsmissingcall: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- INFO_AN: integer (nullable = true)\n",
      " |-- INFO_callsets: integer (nullable = true)\n",
      " |-- INFO_callsetwiththisuniqgenopassing: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- INFO_callable: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- INFO_difficultregion: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- INFO_datasets: integer (nullable = true)\n",
      " |-- INFO_platformbias: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- INFO_platforms: integer (nullable = true)\n",
      " |-- INFO_arbitrated: string (nullable = true)\n",
      " |-- INFO_filt: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- INFO_datasetnames: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- genotypes: array (nullable = true)\n",
      " |    |-- element: struct (containsNull = true)\n",
      " |    |    |-- sampleId: string (nullable = true)\n",
      " |    |    |-- conditionalQuality: integer (nullable = true)\n",
      " |    |    |-- alleleDepths: array (nullable = true)\n",
      " |    |    |    |-- element: integer (containsNull = true)\n",
      " |    |    |-- phased: boolean (nullable = true)\n",
      " |    |    |-- calls: array (nullable = true)\n",
      " |    |    |    |-- element: integer (containsNull = true)\n",
      " |    |    |-- depth: integer (nullable = true)\n",
      " |    |    |-- IGT: string (nullable = true)\n",
      " |    |    |-- IPS: string (nullable = true)\n",
      " |    |    |-- ADALL: array (nullable = true)\n",
      " |    |    |    |-- element: integer (containsNull = true)\n",
      " |    |    |-- PS: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "bcbio_germline_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "# Page Break"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ETL ON NESTED GENOTYPES COLUMN"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "bcbio_germline_df_exploded = bcbio_germline_df.withColumn(\"genotypes\", explode(\"genotypes\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- contigName: string (nullable = true)\n",
      " |-- start: long (nullable = true)\n",
      " |-- end: long (nullable = true)\n",
      " |-- names: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- referenceAllele: string (nullable = true)\n",
      " |-- alternateAlleles: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- qual: double (nullable = true)\n",
      " |-- filters: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- splitFromMultiAllelic: boolean (nullable = true)\n",
      " |-- INFO_platformnames: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- INFO_callsetwithotheruniqgenopassing: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- INFO_callsetnames: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- INFO_AC: array (nullable = true)\n",
      " |    |-- element: integer (containsNull = true)\n",
      " |-- INFO_varType: string (nullable = true)\n",
      " |-- INFO_DPSum: integer (nullable = true)\n",
      " |-- INFO_datasetsmissingcall: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- INFO_AN: integer (nullable = true)\n",
      " |-- INFO_callsets: integer (nullable = true)\n",
      " |-- INFO_callsetwiththisuniqgenopassing: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- INFO_callable: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- INFO_difficultregion: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- INFO_datasets: integer (nullable = true)\n",
      " |-- INFO_platformbias: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- INFO_platforms: integer (nullable = true)\n",
      " |-- INFO_arbitrated: string (nullable = true)\n",
      " |-- INFO_filt: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- INFO_datasetnames: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- genotypes: struct (nullable = true)\n",
      " |    |-- sampleId: string (nullable = true)\n",
      " |    |-- conditionalQuality: integer (nullable = true)\n",
      " |    |-- alleleDepths: array (nullable = true)\n",
      " |    |    |-- element: integer (containsNull = true)\n",
      " |    |-- phased: boolean (nullable = true)\n",
      " |    |-- calls: array (nullable = true)\n",
      " |    |    |-- element: integer (containsNull = true)\n",
      " |    |-- depth: integer (nullable = true)\n",
      " |    |-- IGT: string (nullable = true)\n",
      " |    |-- IPS: string (nullable = true)\n",
      " |    |-- ADALL: array (nullable = true)\n",
      " |    |    |-- element: integer (containsNull = true)\n",
      " |    |-- PS: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "bcbio_germline_df_exploded.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "# Page Break"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## FLATTEN NESTED GENOTYPE FIELDS"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "def flatten_struct_fields(df_):\n",
    "    \"\"\"\n",
    "    REF https://github.com/microsoft/genomicsnotebook\n",
    "    :param df_:\n",
    "    :return:\n",
    "    \"\"\"\n",
    "    flat_cols = [c[0] for c in df_.dtypes if c[1][:6] != 'struct']\n",
    "    nested_cols = [c[0] for c in df_.dtypes if c[1][:6] =='struct']\n",
    "    flat_df = df_.select(flat_cols + [col(nc+'.'+c).alias(nc+'_'+c) for nc in nested_cols for c in df_.select(nc+'.*').columns])\n",
    "    return flat_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "bcbio_germline_df_exploded_flatten = flatten_struct_fields(bcbio_germline_df_exploded)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- contigName: string (nullable = true)\n",
      " |-- start: long (nullable = true)\n",
      " |-- end: long (nullable = true)\n",
      " |-- names: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- referenceAllele: string (nullable = true)\n",
      " |-- alternateAlleles: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- qual: double (nullable = true)\n",
      " |-- filters: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- splitFromMultiAllelic: boolean (nullable = true)\n",
      " |-- INFO_platformnames: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- INFO_callsetwithotheruniqgenopassing: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- INFO_callsetnames: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- INFO_AC: array (nullable = true)\n",
      " |    |-- element: integer (containsNull = true)\n",
      " |-- INFO_varType: string (nullable = true)\n",
      " |-- INFO_DPSum: integer (nullable = true)\n",
      " |-- INFO_datasetsmissingcall: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- INFO_AN: integer (nullable = true)\n",
      " |-- INFO_callsets: integer (nullable = true)\n",
      " |-- INFO_callsetwiththisuniqgenopassing: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- INFO_callable: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- INFO_difficultregion: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- INFO_datasets: integer (nullable = true)\n",
      " |-- INFO_platformbias: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- INFO_platforms: integer (nullable = true)\n",
      " |-- INFO_arbitrated: string (nullable = true)\n",
      " |-- INFO_filt: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- INFO_datasetnames: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- genotypes_sampleId: string (nullable = true)\n",
      " |-- genotypes_conditionalQuality: integer (nullable = true)\n",
      " |-- genotypes_alleleDepths: array (nullable = true)\n",
      " |    |-- element: integer (containsNull = true)\n",
      " |-- genotypes_phased: boolean (nullable = true)\n",
      " |-- genotypes_calls: array (nullable = true)\n",
      " |    |-- element: integer (containsNull = true)\n",
      " |-- genotypes_depth: integer (nullable = true)\n",
      " |-- genotypes_IGT: string (nullable = true)\n",
      " |-- genotypes_IPS: string (nullable = true)\n",
      " |-- genotypes_ADALL: array (nullable = true)\n",
      " |    |-- element: integer (containsNull = true)\n",
      " |-- genotypes_PS: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "bcbio_germline_df_exploded_flatten.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "# Page Break"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## QA TRANSFORMED DATAFRAME"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING\t2022-10-07 23:38:10\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n",
      "WARNING\t2022-10-07 23:38:10\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n",
      "WARNING\t2022-10-07 23:38:10\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n",
      "WARNING\t2022-10-07 23:38:10\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n",
      "[Stage 2:==============>                                            (1 + 3) / 4]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+\n",
      "|genotypes_sampleId|\n",
      "+------------------+\n",
      "|           NA12878|\n",
      "+------------------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "bcbio_germline_df_exploded_flatten.select(\"genotypes_sampleId\").distinct().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+------+------+-------+------------------+---------------+----------------------+\n",
      "|contigName| start|   end|INFO_AC|genotypes_sampleId|genotypes_calls|genotypes_alleleDepths|\n",
      "+----------+------+------+-------+------------------+---------------+----------------------+\n",
      "|      chr1|817185|817186|    [2]|           NA12878|         [0, 1]|             [78, 454]|\n",
      "|      chr1|817340|817341|    [2]|           NA12878|         [0, 1]|            [107, 342]|\n",
      "|      chr1|817888|817889|    [2]|           NA12878|         [0, 1]|             [74, 220]|\n",
      "|      chr1|818801|818802|    [2]|           NA12878|         [0, 1]|              [0, 202]|\n",
      "|      chr1|818811|818812|    [2]|           NA12878|         [0, 1]|              [0, 190]|\n",
      "|      chr1|818953|818954|    [2]|           NA12878|         [0, 1]|              [0, 246]|\n",
      "|      chr1|819122|819123|    [2]|           NA12878|         [0, 1]|            [110, 363]|\n",
      "|      chr1|819583|819584|    [2]|           NA12878|         [0, 1]|             [91, 378]|\n",
      "|      chr1|824319|824320|    [2]|           NA12878|         [0, 1]|             [92, 391]|\n",
      "|      chr1|824456|824457|    [2]|           NA12878|         [0, 1]|              [0, 355]|\n",
      "|      chr1|825531|825532|    [2]|           NA12878|         [0, 1]|              [0, 243]|\n",
      "|      chr1|825766|825767|    [2]|           NA12878|         [0, 1]|              [0, 244]|\n",
      "|      chr1|826576|826577|    [2]|           NA12878|         [0, 1]|                [0, 0]|\n",
      "|      chr1|826892|826893|    [2]|           NA12878|         [0, 1]|             [60, 224]|\n",
      "|      chr1|827208|827209|    [2]|           NA12878|         [0, 1]|              [0, 259]|\n",
      "|      chr1|827211|827212|    [2]|           NA12878|         [0, 1]|              [0, 263]|\n",
      "|      chr1|827220|827221|    [2]|           NA12878|         [0, 1]|              [0, 259]|\n",
      "|      chr1|827251|827252|    [2]|           NA12878|         [0, 1]|              [0, 257]|\n",
      "|      chr1|828013|828014|    [2]|           NA12878|         [0, 1]|                [0, 0]|\n",
      "|      chr1|830724|830725|    [2]|           NA12878|         [0, 1]|              [0, 171]|\n",
      "+----------+------+------+-------+------------------+---------------+----------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING\t2022-10-07 23:38:29\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n"
     ]
    }
   ],
   "source": [
    "bcbio_germline_df_exploded_flatten \\\n",
    "    .select(\"contigName\", \"start\", \"end\", \"INFO_AC\", \"genotypes_sampleId\", \"genotypes_calls\", \"genotypes_alleleDepths\") \\\n",
    "    .where(\"genotypes_sampleId = 'NA12878'\") \\\n",
    "    .show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING\t2022-10-07 23:38:32\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n",
      "WARNING\t2022-10-07 23:38:32\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n",
      "WARNING\t2022-10-07 23:38:32\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n",
      "WARNING\t2022-10-07 23:38:32\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n",
      "[Stage 6:==============>                                            (1 + 3) / 4]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+-------+\n",
      "|     INFO_AC|  count|\n",
      "+------------+-------+\n",
      "|      [2, 0]|   9751|\n",
      "|         [1]|1016830|\n",
      "|   [1, 0, 0]|   1377|\n",
      "|      [1, 0]|  16177|\n",
      "|[1, 1, 0, 0]|    940|\n",
      "|        null|2123774|\n",
      "|      [1, 1]|  14887|\n",
      "|   [2, 0, 0]|    748|\n",
      "|         [2]|1057338|\n",
      "|   [1, 1, 0]|   5726|\n",
      "+------------+-------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "bcbio_germline_df_exploded_flatten \\\n",
    "    .cube(\"INFO_AC\") \\\n",
    "    .count() \\\n",
    "    .show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING\t2022-10-07 23:38:48\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n",
      "WARNING\t2022-10-07 23:38:48\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n",
      "WARNING\t2022-10-07 23:38:48\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n",
      "WARNING\t2022-10-07 23:38:48\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n",
      "[Stage 9:============================================>              (3 + 1) / 4]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-----+\n",
      "|    alternateAlleles|count|\n",
      "+--------------------+-----+\n",
      "|              [TAGC]|   22|\n",
      "|            [GAATGC]|    1|\n",
      "|    [ATGTG, ATGTGTG]|   28|\n",
      "|             [TAAAC]|  141|\n",
      "|         [GTGTTTTCT]|    1|\n",
      "|        [TTTGTTG, T]|    8|\n",
      "|              [GCAA]|   18|\n",
      "|  [CTCTA, CTCTATCTA]|    6|\n",
      "| [TTG, TTGTGTGTG, T]|    2|\n",
      "|          [ATTTCTAT]|    1|\n",
      "|          [C, CGTGT]|   63|\n",
      "|             [TATTC]|   17|\n",
      "|[CATCTATCTATCT, C...|    1|\n",
      "| [CGT, CGTGTGTGT, C]|    2|\n",
      "|[CATT, CATTATT, C...|    3|\n",
      "|          [AAAGTAGC]|    1|\n",
      "|            [ATTTCT]|   14|\n",
      "|     [TAC, T, TACAC]|   15|\n",
      "| [TCACACACACACACACA]|    1|\n",
      "|         [GCCC, GCC]|    1|\n",
      "+--------------------+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "bcbio_germline_df_exploded_flatten \\\n",
    "    .cube(\"alternateAlleles\") \\\n",
    "    .count() \\\n",
    "    .show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING\t2022-10-07 23:39:04\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n",
      "WARNING\t2022-10-07 23:39:04\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n",
      "WARNING\t2022-10-07 23:39:04\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n",
      "WARNING\t2022-10-07 23:39:04\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n",
      "[Stage 12:=============================>                            (2 + 2) / 4]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "number of samples: 1\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "n = bcbio_germline_df_exploded_flatten.select(\"genotypes_sampleId\").distinct().count()\n",
    "print(f\"number of samples: {n}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING\t2022-10-07 23:39:25\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n",
      "WARNING\t2022-10-07 23:39:25\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n",
      "WARNING\t2022-10-07 23:39:25\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n",
      "WARNING\t2022-10-07 23:39:25\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n",
      "[Stage 18:==============>                                           (1 + 3) / 4]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "number of records: 2123774\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "n = bcbio_germline_df_exploded_flatten.count()\n",
    "print(f\"number of records: {n}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING\t2022-10-07 23:39:37\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n",
      "WARNING\t2022-10-07 23:39:37\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n",
      "WARNING\t2022-10-07 23:39:37\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n",
      "WARNING\t2022-10-07 23:39:37\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n",
      "[Stage 21:==============>                                           (1 + 3) / 4]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "number of multiallelic sites: 49606\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "n = bcbio_germline_df_exploded_flatten \\\n",
    "    .where(size(\"alternateAlleles\") > 1) \\\n",
    "    .count()\n",
    "\n",
    "print(f\"number of multiallelic sites: {n}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "# Page Break"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Write to Deltatable"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "germline_table = \"./lakehouse/bcbio/germline_table\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/10/07 23:40:09 WARN package: Truncated the string representation of a plan since it was too large. This behavior can be adjusted by setting 'spark.sql.debug.maxToStringFields'.\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING\t2022-10-07 23:40:11\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n",
      "WARNING\t2022-10-07 23:40:11\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n",
      "WARNING\t2022-10-07 23:40:11\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n",
      "WARNING\t2022-10-07 23:40:12\tAsciiLineReader\tCreating an indexable source for an AsciiFeatureCodec using a stream that is neither a PositionalBufferedStream nor a BlockCompressedInputStream\n",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "bcbio_germline_df_exploded_flatten.write.format(\"delta\").mode(\"overwrite\").save(germline_table)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\u001b[01;34m./lakehouse/bcbio/germline_table\u001b[0m\n",
      "├── \u001b[01;34m_delta_log\u001b[0m\n",
      "│   └── \u001b[00m00000000000000000000.json\u001b[0m\n",
      "├── \u001b[00mpart-00000-e6bec1f6-e168-4966-8792-aae80b7f2fb7-c000.snappy.parquet\u001b[0m\n",
      "├── \u001b[00mpart-00001-243f7c82-8ee4-428b-8a79-0652c674c3cb-c000.snappy.parquet\u001b[0m\n",
      "├── \u001b[00mpart-00002-9b626eac-4afe-499d-8416-348ba0950f6f-c000.snappy.parquet\u001b[0m\n",
      "└── \u001b[00mpart-00003-a2267a27-bc20-47ab-8099-37d25dffc783-c000.snappy.parquet\u001b[0m\n",
      "\n",
      "1 directory, 5 files\n"
     ]
    }
   ],
   "source": [
    "!tree ./lakehouse/bcbio/germline_table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "# Page Break"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Stop Spark Session"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Continue to next notebook"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}