06_metadata_deltatable.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import pyspark\n",
    "from delta import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "builder = pyspark.sql.SparkSession.builder.appName(\"MetadataDeltalakeETL\") \\\n",
    "    .config(\"spark.sql.extensions\", \"io.delta.sql.DeltaSparkSessionExtension\") \\\n",
    "    .config(\"spark.sql.catalog.spark_catalog\", \"org.apache.spark.sql.delta.catalog.DeltaCatalog\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = configure_spark_with_delta_pip(builder).getOrCreate()"
   ]
  },
  {
   "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": [
    "# Load Metadata and Save as Deltatable\n",
    "\n",
    "* NOTE:\n",
    "    * This `meatadata.tsv` is total make up mock for demonstration.\n",
    "    * To simplify thing, it is prepared in denormalized flat table form."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "meta_src = \"./data/metadata.csv\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "meta_df = spark.read.csv(meta_src, header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- SequenceRunName: string (nullable = true)\n",
      " |-- SequenceRunID: string (nullable = true)\n",
      " |-- Timestamp: string (nullable = true)\n",
      " |-- SubjectID: string (nullable = true)\n",
      " |-- LibraryID: string (nullable = true)\n",
      " |-- SampleID: string (nullable = true)\n",
      " |-- SampleDescription: string (nullable = true)\n",
      " |-- Gender: string (nullable = true)\n",
      " |-- Race: string (nullable = true)\n",
      " |-- Ethnicity: string (nullable = true)\n",
      " |-- Proband: string (nullable = true)\n",
      " |-- ProjectOwner: string (nullable = true)\n",
      " |-- ProjectName: string (nullable = true)\n",
      " |-- StudyID: string (nullable = true)\n",
      " |-- StudyType: string (nullable = true)\n",
      " |-- Pipeline: string (nullable = true)\n",
      " |-- Phenotype: string (nullable = true)\n",
      " |-- Collection: string (nullable = true)\n",
      " |-- DiseaseCode: string (nullable = true)\n",
      " |-- SNOMED: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "meta_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "# Page Break"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Data Warehouse Models\n",
    "\n",
    "> Consider: some ETL or complex transformation have done here!! :)\n",
    "\n",
    "* At this point, we can take one step back and think about leveraging Data Warehouse data models.\n",
    " * Some examples but not limited to:\n",
    "    * Multi-Dimensional data model such as Star schema, Snowflake schema\n",
    "        * For example, wrap around our `somatic_table` as Fact table and, build metadata as surrounding dimension table(s)\n",
    "    * Data Vault data modelling such as arranging in Hub, Satellite, Links concepts (depends on data stages i.e. Bronze, Silver, Gold, Platinum)\n",
    "    * Or, simple flat table, Data Mart or, even some simplified Relation model if that suit for the use case"
   ]
  },
  {
   "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": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|SampleID|\n",
      "+--------+\n",
      "| NA24385|\n",
      "| NA12878|\n",
      "+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "meta_df.select(\"SampleID\").distinct().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+----------+---------+---------+--------+-------+---------+--------+-------------+---------+\n",
      "|     SequenceRunName| Timestamp|SubjectID|LibraryID|SampleID|StudyID|StudyType|Pipeline|  DiseaseCode|   SNOMED|\n",
      "+--------------------+----------+---------+---------+--------+-------+---------+--------+-------------+---------+\n",
      "|221007_A00130_000...|2022-10-07| SBJ00001| L0000001| NA12878|NA12878|      WGS| Somatic|MONDO:0007254|429740004|\n",
      "+--------------------+----------+---------+---------+--------+-------+---------+--------+-------------+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "meta_df \\\n",
    "    .select(\"SequenceRunName\", \"Timestamp\", \"SubjectID\", \"LibraryID\", \"SampleID\", \"StudyID\", \"StudyType\", \"Pipeline\", \"DiseaseCode\", \"SNOMED\") \\\n",
    "    .where(\"SampleID = 'NA12878'\") \\\n",
    "    .show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-----+\n",
      "|StudyID|count|\n",
      "+-------+-----+\n",
      "|NA12878|    2|\n",
      "|   null|    2|\n",
      "+-------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "meta_df \\\n",
    "    .dropna() \\\n",
    "    .cube(\"StudyID\") \\\n",
    "    .count() \\\n",
    "    .show()"
   ]
  },
  {
   "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": 10,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "metadata_table = \"./lakehouse/bcbio/metadata_table\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "meta_df.write.format(\"delta\").mode(\"overwrite\").save(metadata_table)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\u001b[1m\u001b[36m_delta_log\u001b[m\u001b[m\n",
      "part-00000-53abd47d-e378-4144-bd28-ba6c00b531a7-c000.snappy.parquet\n"
     ]
    }
   ],
   "source": [
    "!ls ./lakehouse/bcbio/metadata_table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Page Break"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Summary\n",
    "\n",
    "* In real world scenario, this could involve much more complex structure and/or integration with upstream systems.\n",
    "    * For example, interfacing with systems such as RedCAP, FHIR, or Pathling, ontology and some kind of clinical Phenotype look up server\n",
    "    * And there may be possibility with dynamically looking them up from these systems\n",
    "* Consider: we need to ingest some minimal metadata info and, need to ETL out these minimal Phenotype info from these systems\n",
    "* We may or may not need to ingest these meta information into data warehouse -- which depends on case by case basis."
   ]
  },
  {
   "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": 13,
   "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
}