04_glow_vcf_deltatable_somatic_databricks.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "3cf9008b-9d69-4d89-a6ad-880b4a4ee28e",
     "showTitle": false,
     "title": ""
    },
    "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": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "5f9db9e9-757f-482d-8054-93ecd9160683",
     "showTitle": false,
     "title": ""
    },
    "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": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "4ae1e35a-7643-4b3f-adc2-c5f3ff784180",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# spark = configure_spark_with_delta_pip(builder, extra_packages=extra_packages).getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "bf16af07-f56f-43c6-bb2b-62675bdf8744",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# spark = glow.register(spark)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "adba7fd4-8d79-4030-95f9-5325d02b725e",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "1435bd96-2e0a-4340-b722-1f13d6bf754b",
     "showTitle": false,
     "title": ""
    },
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "# Page Break"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "851c8236-eb66-4210-aaed-150011f149d1",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "# ETL Somatic VCF and Save as Deltatable"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "2679208e-051b-4809-b99a-b99b84afe0e5",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "bcbio_somatic_src = \"dbfs:/tmp/dannywong/na12878-na24385-somatic-hg38-truth.vcf.gz\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "3a3f95cd-9a5c-4a7c-8e41-5068f1e10015",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "bcbio_somatic_df = spark.read.format(\"vcf\").load(bcbio_somatic_src)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "690b42c2-686b-43d1-a2ca-38481a9bee40",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "bcbio_somatic_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "b95fcedc-ce4c-4033-8fb8-9c98204c6583",
     "showTitle": false,
     "title": ""
    },
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "# Page Break"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "369c7a03-d7f7-411e-b4ed-083ec6afa4ae",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "## ETL ON NESTED GENOTYPES COLUMN"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "4ca8ad1e-cfcd-457f-bddd-99ef56c9b4d3",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "bcbio_somatic_df_exploded = bcbio_somatic_df.withColumn(\"genotypes\", explode(\"genotypes\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "1859083e-3388-455b-86f3-f1b2906a2f42",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "bcbio_somatic_df_exploded.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "a3916712-a05c-4454-a717-00a6da64cf72",
     "showTitle": false,
     "title": ""
    },
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "# Page Break"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "af06671c-417d-4f5c-a4aa-a219b443e2ec",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "## FLATTEN NESTED GENOTYPE FIELDS"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "1714279e-09ec-4bb6-9a71-36bcc852dfdb",
     "showTitle": false,
     "title": ""
    }
   },
   "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": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "5cb74493-a5d7-4322-a573-2cce843d2b35",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "bcbio_somatic_df_exploded_flatten = flatten_struct_fields(bcbio_somatic_df_exploded)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "e928157a-df40-49ff-a253-b884b0b2f1af",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "bcbio_somatic_df_exploded_flatten.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "e74e0b39-ac49-4278-b899-e744c7c37fc8",
     "showTitle": false,
     "title": ""
    },
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "# Page Break"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "33b64553-c9ff-48ce-a9f9-e89b603ba74b",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "## QA TRANSFORMED DATAFRAME"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "1b678255-4e0a-412b-951e-dec7b17875d7",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "bcbio_somatic_df_exploded_flatten.select(\"genotypes_sampleId\").distinct().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "cf81b0e7-f899-4991-9089-735ace3e39e4",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "bcbio_somatic_df_exploded_flatten \\\n",
    "    .select(\"contigName\", \"start\", \"end\", \"INFO_AC\", \"INFO_SOMTYPE\", \"genotypes_sampleId\", \"genotypes_calls\", \"genotypes_alleleDepths\") \\\n",
    "    .where(\"genotypes_sampleId = 'NA12878'\") \\\n",
    "    .show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "e9979e90-051e-4edc-b445-4597fda84a1c",
     "showTitle": false,
     "title": ""
    },
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "bcbio_somatic_df_exploded_flatten \\\n",
    "    .cube(\"INFO_AC\") \\\n",
    "    .count() \\\n",
    "    .show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "4ecb4f2f-e201-43ea-8b82-74698199c9ee",
     "showTitle": false,
     "title": ""
    },
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "bcbio_somatic_df_exploded_flatten \\\n",
    "    .cube(\"alternateAlleles\") \\\n",
    "    .count() \\\n",
    "    .show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "251c5a0c-f268-4a14-8f3b-9cff50d3d79a",
     "showTitle": false,
     "title": ""
    },
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "n = bcbio_somatic_df_exploded_flatten.select(\"genotypes_sampleId\").distinct().count()\n",
    "print(f\"number of samples: {n}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "110e0ce9-d7a7-470a-9098-e2351743f550",
     "showTitle": false,
     "title": ""
    },
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "n = bcbio_somatic_df_exploded_flatten.count()\n",
    "print(f\"number of records: {n}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "d23cd17e-d04e-4d33-a672-0647381df0b0",
     "showTitle": false,
     "title": ""
    },
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "n = bcbio_somatic_df_exploded_flatten \\\n",
    "    .where(size(\"alternateAlleles\") > 1) \\\n",
    "    .count()\n",
    "\n",
    "print(f\"number of multiallelic sites: {n}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "b22a708e-beb3-49af-94f1-6b946509caf5",
     "showTitle": false,
     "title": ""
    },
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "# Page Break"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "c01a7910-dcc6-49ad-aa62-e75ae1d9bc65",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "# Write to Deltatable"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "68aa8694-8c56-4b7f-a29d-0a5f06eeaa33",
     "showTitle": false,
     "title": ""
    },
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "# somatic_table = \"./lakehouse/bcbio/somatic_table\"\n",
    "somatic_table = \"dbfs:/tmp/dannywong/lakehouse/bcbio/somatic_table\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "74768352-340c-4415-852d-8597994a929e",
     "showTitle": false,
     "title": ""
    },
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "bcbio_somatic_df_exploded_flatten.write.format(\"delta\").mode(\"overwrite\").save(somatic_table)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "6987d1bf-56e6-47e8-9bfe-d6aa185af513",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "%fs\n",
    "ls dbfs:/tmp/dannywong/lakehouse/bcbio/somatic_table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "d886f432-e469-41f5-af12-3e26186e5a5b",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "!tree ./lakehouse/bcbio/somatic_table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "18126efe-6e76-4e0a-9e2e-6692cb2a37ad",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# Page Break"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "edc048e8-8970-4934-be28-e8989ca78aef",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "# Summary\n",
    "\n",
    "* In this example, we read VCF through Spark/Glow.\n",
    "* We perform data transformation on Genotype column.\n",
    "* We flatten (i.e. data normalisation) the nested vectors of Genotype column.\n",
    "* So that it can be performant and ease of use at data query (i.e. trade off for information retrieval time) by its predicates such as Sample ID.\n",
    "* We then write this dataframe out as multi-parts compressed Parquet files through Deltatable framework.\n",
    "* We can write Spark dataframe in multiple Deltatable write modes such as:\n",
    "    * We would use `upsert` mode when/if we were to update or insert new records into existing delta table.\n",
    "    * Or, `append` mode, if destination table is immutable and, so on so ford.\n",
    "* If we wish, we could also write as-is Parquet only format and arrange them in traditional Datalake structure.\n",
    "    * e.g. `bcbio_somatic_df_exploded_flatten.write.format(\"parquet\").mode(\"overwrite\").save(\"./datalake/pipeline=bcbio/type=somatic/year=2022/month=01\")`\n",
    "    * In this case, we will be responsible for maintaining Datalake structure and its key-value partitioning such as `pipeline=bcbio`, etc\n",
    "* With Deltatable framework, we abstract away this and leave it up to Deltatable framework; and achieve \"Logical Table\" that perform like relational database table.\n",
    "* Hence, Deltatable as such technology underpin \"LakeHouse\" architecture pattern for BigData data warehousing possibility."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "34329660-04d7-4763-a138-8fff46eb6621",
     "showTitle": false,
     "title": ""
    },
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "# Page Break"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "ec710ab4-31b4-465d-bafe-b56d5121c1cf",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "# Stop Spark Session"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "913f2099-1f60-4ef0-b3fd-4665dd6f5e5a",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# spark.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "139bcac8-5d68-491d-b8c1-dc79132d4d9b",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# Continue to next notebook"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "6d8654cc-8cd4-4b02-9ff5-2fb4bd21bb7e",
     "showTitle": false,
     "title": ""
    },
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "application/vnd.databricks.v1+notebook": {
   "dashboards": [],
   "language": "python",
   "notebookMetadata": {
    "mostRecentlyExecutedCommandWithImplicitDF": {
     "commandId": 649139525341151,
     "dataframes": [
      "_sqldf"
     ]
    },
    "pythonIndentUnit": 2
   },
   "notebookName": "04_glow_vcf_deltatable_somatic_databricks",
   "notebookOrigID": 649139525341011,
   "widgets": {}
  },
  "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": 0
}