Experiments.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "import time\n",
    "\n",
    "# New API\n",
    "spark_session = SparkSession\\\n",
    "        .builder\\\n",
    "        .master(\"spark://group101:7077\") \\\n",
    "        .appName(\"Average_protein\")\\\n",
    "        .getOrCreate()\n",
    "\n",
    "# Old API (RDD)\n",
    "spark_context = spark_session.sparkContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Start taking time\n",
    "start_time = time.time()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# function that \n",
    "def count_occur(str):\n",
    "   uniq = set(str)\n",
    "   li = list(str)\n",
    "   dict = {}\n",
    "   for key in uniq:\n",
    "       dict[key] = str.count(key)\n",
    "   return dict"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "def count_all(str):\n",
    "   li = list(str)\n",
    "   length = len(li)\n",
    "   return length"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "def add(dict1, dict2):\n",
    "    result = {key: dict1.get(key, 0) + dict2.get(key, 0)\n",
    "          for key in set(dict1) | set(dict2)}\n",
    "    return result"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "fasta_file = spark_context.textFile('file:///home/ubuntu/data/pdb_seqres4.txt')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Split the fasta file into headers and amino acid sequences\n",
    "headers = fasta_file.filter(lambda x: \">\" in x)\n",
    "sequences = fasta_file.filter(lambda x: \">\" not in x)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "headers_indexed = headers.zipWithIndex()\n",
    "sequences_indexed = sequences.zipWithIndex()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "sequences_indexed_inverted = sequences_indexed.map(lambda x: (x[1], x[0]))\n",
    "headers_indexed_inverted = headers_indexed.map(lambda x: (x[1], x[0]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Joining the sequences by their indexes\n",
    "joined_fasta = sequences_indexed_inverted.join(headers_indexed_inverted)\n",
    "# Filter out all non-proteins\n",
    "prot_sequences = joined_fasta.filter(lambda x: (\"protein\" in x[1][1]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Counting the occurence of each character in each protein\n",
    "ind_dict = prot_sequences.map(lambda x: count_occur(x[1][0]))\n",
    "# Taking a look at two first proteins\n",
    "# ind_dict.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{'P': 0.046321832081362375, 'K': 0.05960191786714657, 'C': 0.013337454860752107, 'D': 0.05555970162922566, 'A': 0.07992560654223138, 'I': 0.05570616612050634, 'R': 0.05269718196823905, 'L': 0.08940539061097663, 'E': 0.06572919486526313, 'H': 0.026479981958879073, 'Z': 5.046310552642776e-07, 'S': 0.06247436193888671, 'M': 0.023248343371005724, 'W': 0.013170552811733217, 'Y': 0.03393936345763929, 'F': 0.03854759818419542, 'Q': 0.037628506167226954, 'G': 0.07387441669322521, 'V': 0.07016072461730088, 'U': 1.5325832048766948e-06, 'N': 0.04170209363575013, 'B': 3.0838564488372514e-07, 'O': 1.869003908386213e-08, 'X': 0.004952608041695832, 'T': 0.0555346382868142}\n"
     ]
    }
   ],
   "source": [
    "sum_lengths = prot_sequences.map(lambda x: count_all(x[1][0])).sum()\n",
    "all_aa_occ = ind_dict.reduce(add)\n",
    "result = {key: (all_aa_occ.get(key, 0))/sum_lengths\n",
    "          for key in set(all_aa_occ)}\n",
    "print(result)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "execution_time = time.time() - start_time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "168.09951519966125\n"
     ]
    }
   ],
   "source": [
    "print(execution_time)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "# release the cores for another application!\n",
    "spark_context.stop()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.5.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}