04_create_datalake_and_athena_tables.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Tutorial 4: Athena tables from pipeline outputs & sample-level metadata in Quilt packages\n",
    "\n",
    "To enable valuable data searches, we must align the sample-level metadata appended to the raw data packages to the pipeline outputs. In this demo, the primary data generated by the pipeline is expression tables. With Athena, its possible to integrate sample metadata & pipeline output tables together to empower quick queries and slicing and dicing of large datasets.\n",
    "\n",
    "Now that we have raw data pacakaged with sample-level metadata, and Nextflow pipeline outputs packaged with `nf-quilt`, the next step is use this data to create tables and views in Athena to enable snappy data queries. \n",
    "\n",
    "For more more information on querying Quilt package metadata in Athena, please refer to the Quilt documentation at <https://docs.quiltdata.com/advanced/athena>.\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import boto3 as boto3"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 1. Connect to Athena"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# user to fill in AWS credentials (or auth in preferred method)\n",
    "aws_access_key_id = \"USER_ACCESS KEY\"\n",
    "aws_secret_access_key = \"USER SECRET ACCESS KEY\"\n",
    "region_name = \"us-east-1\"\n",
    "bucket_name = \"s3://quilt-example-bucket/ccle\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Connect to Athena\n",
    "conn = connect(aws_access_key_id=aws_access_key_id,\n",
    "               aws_secret_access_key=aws_secret_access_key,\n",
    "               region_name=region_name,\n",
    "               s3_staging_dir= bucket_name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 2. Generate symlink datalake across Nextflow pipeline runs\n",
    "\n",
    "Typically Athena and AWS Glue prefer data to be organized in a specific structure, usually 1 file per directory, where the directory levels represent a partition key. `nf-core` Nextflow pipelines do not output data in the traditionally Athena compatible structure, with many files per directory and directories often representing the names of different tools/steps in the pipeline. \n",
    "\n",
    "As an alternative to re-structuring or copying pipeline outputs to be compatible with Athena organization, in this demo we use symbolic links (symlinks) to create a datalake of Nextflow pipeline outputs partitioned by `run_id` and `sample_id`. We then use this data lake to create a view representing a table of each samples gene expression values from a `quant.sf` pipeline output across all samples across all runs. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "s3 = boto3.resource('s3')\n",
    "bucket_name = \"quilt-example-bucket\"\n",
    "prefix = \"ccle/datalake\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# define the quant.sf files we want to use to populate the datalake\n",
    "# a similar method can be used for other piepline outputs\n",
    "parse_keys = \"\"\"\n",
    "WITH extracted_data AS (\n",
    "    SELECT\n",
    "        timestamp,\n",
    "        logical_key,\n",
    "        regexp_extract(logical_key, '^[^/]+/([^/]+)/.*', 1) AS sample_id,\n",
    "        physical_key,\n",
    "        regexp_extract(physical_key, '^(.*?)\\?versionId=', 1) AS object_path\n",
    "    FROM\n",
    "        \"userathenadatabase-2htmlbiqyvry\".\"quilt-example-bucket_objects-view\"\n",
    "    WHERE\n",
    "        logical_key LIKE '%quant.sf'\n",
    ")\n",
    "SELECT\n",
    "    logical_key,\n",
    "    sample_id,\n",
    "    physical_key,\n",
    "    object_path\n",
    "FROM\n",
    "    extracted_data\n",
    "WHERE\n",
    "    sample_id LIKE 'SRR%';\n",
    "    AND timestamp = 'latest';\n",
    "\"\"\"\n",
    "\n",
    "cursor = conn.cursor()\n",
    "cursor.execute(parse_keys)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# generate a symlinks file for each file of interest\n",
    "for row in cursor:\n",
    "        sample_id = row[1]\n",
    "        run_id = row[3].split(\"/\")[4]\n",
    "        object_path = row[3]\n",
    "        s3.Bucket(bucket_name).put_object(Key=f\"{prefix}/run_id={run_id}/sample_id={sample_id}/symlink.txt\", Body=object_path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# use the symlink data lake to create a new table containins samples x gene expression across all runs\n",
    "parse_keys = \"\"\"\n",
    "CREATE EXTERNAL TABLE `ccle_nfcore_rnaseq_datalake`(\n",
    "  `name` string COMMENT 'from deserializer', \n",
    "  `length` string COMMENT 'from deserializer', \n",
    "  `effectivelength` string COMMENT 'from deserializer', \n",
    "  `tpm` string COMMENT 'from deserializer', \n",
    "  `numreads` string COMMENT 'from deserializer')\n",
    "ROW FORMAT SERDE \n",
    "  'org.apache.hadoop.hive.serde2.OpenCSVSerde' \n",
    "WITH SERDEPROPERTIES ( \n",
    "  'separatorChar'='\\t') \n",
    "STORED AS INPUTFORMAT \n",
    "  'org.apache.hadoop.mapred.TextInputFormat' \n",
    "OUTPUTFORMAT \n",
    "  'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'\n",
    "LOCATION\n",
    "  's3://quilt-example-bucket/ccle/datalake/'\n",
    "  tblproperties (\n",
    "  'skip.header.line.count'='1'\n",
    ")\n",
    "\"\"\"\n",
    "cursor = conn.cursor()\n",
    "cursor.execute(parse_keys)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 3. Create table for sample-level metadata\n",
    "\n",
    "Quilt package metadata is stored on s3 as a json file. We can use this metadata to create a table in Athena. Below is example SQL code to create this table view in Athena , which will output a sample x field table for all the sample-level metadata appended to raw data packages in Quilt. \n",
    "\n",
    "\n",
    "```sql\n",
    "--- create expanded view of raw pkg metadata\n",
    "CREATE OR REPLACE VIEW ccle_raw_data_pkg_metadata AS\n",
    "SELECT\n",
    "    *,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.lineage') AS lineage,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.createdate') AS createdate,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.run')AS run,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.tissue')AS tissue,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.primarytumorsite') AS primarytumorsite,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.releasedate')AS releasedate,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.mutationrate')AS mutationrate,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.cellline') AS cellline,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.celllinenickname') AS celllinenickname,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.oncotreecode')AS oncotreecode,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.experiment') AS experiment,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.histology') AS histology,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.tmbnonsynonymous') AS tmbnonsynonymous,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.avgspotlen') AS avgspotlen,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.datastoreprovider') AS datastoreprovider,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.srastudy') AS srastudy,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.biosample') AS biosample,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.pathologistannotation') AS pathologistannotation,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.samplename') AS samplename,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.doublingtimehrs') AS doublingtimehrs,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.organism') AS organism,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.genomedoublings') AS genomedoublings,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.biosamplemodel') AS biosamplemodel,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.sitesubtype1') AS sitesubtype1,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.sitesubtype2') AS sitesubtype2,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.datastorefiletype') AS datastorefiletype,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.datastoreregion') AS datastoreregion,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.fractiongenomealtered') AS fractiongenomealtered,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.consent') AS consent,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.centername') AS centername,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.ploidy') AS ploidy,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.freezingmedium')AS freezingmedium,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.siteoffinding') AS siteoffinding,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.growthmedium') AS growthmedium,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.annotationsource') AS annotationsource,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.ethnicity') AS ethnicity,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.assemblyname') AS assemblyname,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.biomaterialprovider') AS biomaterialprovider,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.purity') AS purity,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.instrument') AS instrument,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.cancertypedetailed') AS cancertypedetailed,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.platform') AS platform,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.cancertype') AS cancertype,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.lineagesubtype') AS lineagesubtype,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.bases') AS bases,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.flowcellid') AS flowcellid,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.diseasestage') AS diseasestage,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.isolate') AS isolate,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.assaytype') AS assaytype,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.histologysubtype1') AS histologysubtype1,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.subtype') AS subtype,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.createdatebatch') AS createdatebatch,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.histologysubtype2') AS histologysubtype2,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.studyid') AS studyid,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.depmapid') AS depmapid,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.librarylayout') AS librarylayout,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.characteristics') AS characteristics,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.disease') AS disease,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.patientid') AS patientid,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.sex') AS sex,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.celllinesource') AS celllinesource,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.librarysource') AS librarysource,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.libraryselection') AS libraryselection,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.bioproject') AS bioproject,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.libraryname') AS libraryname,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.sampletype') AS sampletype,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.mutationcount') AS mutationcount,\n",
    "    JSON_EXTRACT_SCALAR(user_meta, '$.age') AS age\n",
    "FROM\n",
    "    \"userathenadatabase-2htmlbiqyvry\".\"quilt-example-bucket_packages-view\"\n",
    "WHERE \n",
    "    substr(pkg_name, -10, 3) = 'SRR'\n",
    "    AND timestamp = 'latest';\n",
    "\n",
    "--- preview new view\n",
    "SELECT * FROM \"userathenadatabase-2htmlbiqyvry\".\"ccle_raw_data_pkg_metadata\" limit 10;\n",
    "\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we can use sample ID as the join key to join these sample-level metadata directly to nf-core pipeline tables created from the symlink datalake."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# "
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "base",
   "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.11.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}