tutorials/snark/Snark Tutorial.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Snark (Snorkel on Spark) Tutorial\n",
    "\n",
    "## Introduction\n",
    "\n",
    "The central data-intensive operation in Snorkel is applying _labeling functions_ to unlabeled data. Snorkel then denoises the outputs of these labeling functions to infer a probabilistic label for each data point and train an end model.\n",
    "\n",
    "We can accelerate the iterative process of developing labeling functions using Spark. The `SparkLabelAnnotator` distributes candidate sets (data points to be labeled) to a Spark cluster, then applies labeling functions and collects the results. We save time by distributing and caching the each candidate set once, then only transmitting labeling functions and collecting labels back in the Snorkel database.\n",
    "\n",
    "This will demonstrate (one way) how to fit Spark into a Snorkel pipeline. This example can be inserted into many other applications.\n",
    "\n",
    "## Installing `PySpark`\n",
    "\n",
    "Please see the [official instructions](https://spark.apache.org/docs/latest/spark-standalone.html)!\n",
    "\n",
    "## Setup\n",
    "In this tutorial, we will continue with the chemical-disease relation extraction task from the [Chemical-Disease Tutorial](https://github.com/HazyResearch/snorkel/tree/master/tutorials/cdr). We will pick up here after the first notebook, the data preprocessing stage, so **remember to run [notebook 1](https://github.com/HazyResearch/snorkel/blob/master/tutorials/cdr/CDR_Tutorial_1.ipynb)** first, or un-zip the pre-run snorkel database from this preprocessing step:\n",
    "```bash\n",
    "cd tutorials/cdr\n",
    "tar -xzvf tar -xzvf snorkel.CDR_1.db.tar.gz\n",
    "mv snorkel.CDR_1.db snorkel.db\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Connecting Snorkel and Spark\n",
    "\n",
    "We first create a `SnorkelSession` to load our data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%load_ext autoreload\n",
    "%autoreload 2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "\n",
    "# Remember to specify the SNORKELDB environment variable (with an absolute path if using the file system)\n",
    "os.environ['SNORKELDB'] = 'sqlite:///' + os.getcwd() + os.sep + '../cdr/snorkel.db'\n",
    "\n",
    "from snorkel import SnorkelSession\n",
    "\n",
    "snorkel_session = SnorkelSession()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We also create a `SparkSession` to connect to a cluster."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "# Specify the URL to your Spark master node here\n",
    "spark_session = SparkSession.builder.master(\"local\").appName(\"Snark Tutorial\").getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Finally, we specify the type of `Candidate` we will label. (This is initially defined in the [Chemical-Disease Tutorial](https://github.com/HazyResearch/snorkel/tree/master/tutorials/cdr).)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from snorkel.models import candidate_subclass\n",
    "\n",
    "ChemicalDisease = candidate_subclass('ChemicalDisease', ['chemical', 'disease'])\n",
    "\n",
    "for k in range(3):\n",
    "    print(\"Number of candidates for split %d: %d\" % (k, snorkel_session.query(ChemicalDisease).filter(ChemicalDisease.split == k).count()))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Writing Labeling Functions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Distant supervision approaches\n",
    "\n",
    "We'll use the [Comparative Toxicogenomics Database](http://ctdbase.org/) (CTD) for distant supervision. The CTD lists chemical-condition entity pairs under three categories: therapy, marker, and unspecified. Therapy means the chemical treats the condition, marker means the chemical is typically present with the condition, and unspecified is... unspecified. We can write LFs based on these categories.\n",
    "\n",
    "Note that we'll load the database as in the [Chemical-Disease Tutorial](https://github.com/HazyResearch/snorkel/tree/master/tutorials/cdr), but now that we're using Spark, we'll also _broadcast_ it to our workers. This allows the distributed labeling functions to use the data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import bz2\n",
    "import cPickle\n",
    "\n",
    "with bz2.BZ2File('../cdr/data/ctd.pkl.bz2', 'rb') as ctd_f:\n",
    "    ctd_unspecified, ctd_therapy, ctd_marker = cPickle.load(ctd_f)\n",
    "\n",
    "ctd_unspecified = spark_session.sparkContext.broadcast(ctd_unspecified)\n",
    "ctd_therapy = spark_session.sparkContext.broadcast(ctd_therapy)\n",
    "ctd_marker = spark_session.sparkContext.broadcast(ctd_marker)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "(We define these primitive membership checks for later reuse.)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def cand_in_ctd_unspecified(c):\n",
    "    return 1 if c.get_cids() in ctd_unspecified.value else 0\n",
    "\n",
    "def cand_in_ctd_therapy(c):\n",
    "    return 1 if c.get_cids() in ctd_therapy.value else 0\n",
    "\n",
    "def cand_in_ctd_marker(c):\n",
    "    return 1 if c.get_cids() in ctd_marker.value else 0"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Distant supervision labeling functions:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def LF_in_ctd_unspecified(c):\n",
    "    return -1 * cand_in_ctd_unspecified(c)\n",
    "\n",
    "def LF_in_ctd_therapy(c):\n",
    "    return -1 * cand_in_ctd_therapy(c)\n",
    "\n",
    "def LF_in_ctd_marker(c):\n",
    "    return cand_in_ctd_marker(c)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Text pattern approaches\n",
    "\n",
    "Now we'll use some LF helpers to create LFs based on indicative text patterns. We came up with these rules by using the viewer to examine training candidates and noting frequent patterns."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import re\n",
    "from snorkel.lf_helpers import (\n",
    "    get_tagged_text,\n",
    "    rule_regex_search_tagged_text,\n",
    "    rule_regex_search_btw_AB,\n",
    "    rule_regex_search_btw_BA,\n",
    "    rule_regex_search_before_A,\n",
    "    rule_regex_search_before_B,\n",
    ")\n",
    "\n",
    "# List to parenthetical\n",
    "def ltp(x):\n",
    "    return '(' + '|'.join(x) + ')'\n",
    "\n",
    "def LF_induce(c):\n",
    "    return 1 if re.search(r'{{A}}.{0,20}induc.{0,20}{{B}}', get_tagged_text(c), flags=re.I) else 0\n",
    "\n",
    "causal_past = ['induced', 'caused', 'due']\n",
    "def LF_d_induced_by_c(c):\n",
    "    return rule_regex_search_btw_BA(c, '.{0,50}' + ltp(causal_past) + '.{0,9}(by|to).{0,50}', 1)\n",
    "def LF_d_induced_by_c_tight(c):\n",
    "    return rule_regex_search_btw_BA(c, '.{0,50}' + ltp(causal_past) + ' (by|to) ', 1)\n",
    "\n",
    "def LF_induce_name(c):\n",
    "    return 1 if 'induc' in c.chemical.get_span().lower() else 0     \n",
    "\n",
    "causal = ['cause[sd]?', 'induce[sd]?', 'associated with']\n",
    "def LF_c_cause_d(c):\n",
    "    return 1 if (\n",
    "        re.search(r'{{A}}.{0,50} ' + ltp(causal) + '.{0,50}{{B}}', get_tagged_text(c), re.I)\n",
    "        and not re.search('{{A}}.{0,50}(not|no).{0,20}' + ltp(causal) + '.{0,50}{{B}}', get_tagged_text(c), re.I)\n",
    "    ) else 0\n",
    "\n",
    "treat = ['treat', 'effective', 'prevent', 'resistant', 'slow', 'promise', 'therap']\n",
    "def LF_d_treat_c(c):\n",
    "    return rule_regex_search_btw_BA(c, '.{0,50}' + ltp(treat) + '.{0,50}', -1)\n",
    "def LF_c_treat_d(c):\n",
    "    return rule_regex_search_btw_AB(c, '.{0,50}' + ltp(treat) + '.{0,50}', -1)\n",
    "def LF_treat_d(c):\n",
    "    return rule_regex_search_before_B(c, ltp(treat) + '.{0,50}', -1)\n",
    "def LF_c_treat_d_wide(c):\n",
    "    return rule_regex_search_btw_AB(c, '.{0,200}' + ltp(treat) + '.{0,200}', -1)\n",
    "\n",
    "def LF_c_d(c):\n",
    "    return 1 if ('{{A}} {{B}}' in get_tagged_text(c)) else 0\n",
    "\n",
    "def LF_c_induced_d(c):\n",
    "    return 1 if (\n",
    "        ('{{A}} {{B}}' in get_tagged_text(c)) and \n",
    "        (('-induc' in c[0].get_span().lower()) or ('-assoc' in c[0].get_span().lower()))\n",
    "        ) else 0\n",
    "\n",
    "def LF_improve_before_disease(c):\n",
    "    return rule_regex_search_before_B(c, 'improv.*', -1)\n",
    "\n",
    "pat_terms = ['in a patient with ', 'in patients with']\n",
    "def LF_in_patient_with(c):\n",
    "    return -1 if re.search(ltp(pat_terms) + '{{B}}', get_tagged_text(c), flags=re.I) else 0\n",
    "\n",
    "uncertain = ['combin', 'possible', 'unlikely']\n",
    "def LF_uncertain(c):\n",
    "    return rule_regex_search_before_A(c, ltp(uncertain) + '.*', -1)\n",
    "\n",
    "def LF_induced_other(c):\n",
    "    return rule_regex_search_tagged_text(c, '{{A}}.{20,1000}-induced {{B}}', -1)\n",
    "\n",
    "def LF_far_c_d(c):\n",
    "    return rule_regex_search_btw_AB(c, '.{100,5000}', -1)\n",
    "\n",
    "def LF_far_d_c(c):\n",
    "    return rule_regex_search_btw_BA(c, '.{100,5000}', -1)\n",
    "\n",
    "def LF_risk_d(c):\n",
    "    return rule_regex_search_before_B(c, 'risk of ', 1)\n",
    "\n",
    "def LF_develop_d_following_c(c):\n",
    "    return 1 if re.search(r'develop.{0,25}{{B}}.{0,25}following.{0,25}{{A}}', get_tagged_text(c), flags=re.I) else 0\n",
    "\n",
    "procedure, following = ['inject', 'administrat'], ['following']\n",
    "def LF_d_following_c(c):\n",
    "    return 1 if re.search('{{B}}.{0,50}' + ltp(following) + '.{0,20}{{A}}.{0,50}' + ltp(procedure), get_tagged_text(c), flags=re.I) else 0\n",
    "\n",
    "def LF_measure(c):\n",
    "    return -1 if re.search('measur.{0,75}{{A}}', get_tagged_text(c), flags=re.I) else 0\n",
    "\n",
    "def LF_level(c):\n",
    "    return -1 if re.search('{{A}}.{0,25} level', get_tagged_text(c), flags=re.I) else 0\n",
    "\n",
    "def LF_neg_d(c):\n",
    "    return -1 if re.search('(none|not|no) .{0,25}{{B}}', get_tagged_text(c), flags=re.I) else 0\n",
    "\n",
    "WEAK_PHRASES = ['none', 'although', 'was carried out', 'was conducted',\n",
    "                'seems', 'suggests', 'risk', 'implicated',\n",
    "               'the aim', 'to (investigate|assess|study)']\n",
    "\n",
    "WEAK_RGX = r'|'.join(WEAK_PHRASES)\n",
    "\n",
    "def LF_weak_assertions(c):\n",
    "    return -1 if re.search(WEAK_RGX, get_tagged_text(c), flags=re.I) else 0"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Composite LFs\n",
    "\n",
    "The following LFs take some of the strongest distant supervision and text pattern LFs, and combine them to form more specific LFs. These LFs introduce some obvious dependencies within the LF set, which we will model later."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def LF_ctd_marker_c_d(c):\n",
    "    return LF_c_d(c) * cand_in_ctd_marker(c)\n",
    "\n",
    "def LF_ctd_marker_induce(c):\n",
    "    return (LF_c_induced_d(c) or LF_d_induced_by_c_tight(c)) * cand_in_ctd_marker(c)\n",
    "\n",
    "def LF_ctd_therapy_treat(c):\n",
    "    return LF_c_treat_d_wide(c) * cand_in_ctd_therapy(c)\n",
    "\n",
    "def LF_ctd_unspecified_treat(c):\n",
    "    return LF_c_treat_d_wide(c) * cand_in_ctd_unspecified(c)\n",
    "\n",
    "def LF_ctd_unspecified_induce(c):\n",
    "    return (LF_c_induced_d(c) or LF_d_induced_by_c_tight(c)) * cand_in_ctd_unspecified(c)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Rules based on context hierarchy\n",
    "\n",
    "These last two rules will make use of the context hierarchy. The first checks if there is a chemical mention much closer to the candidate's disease mention than the candidate's chemical mention. The second does the analog for diseases."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def LF_closer_chem(c):\n",
    "    # Get distance between chemical and disease\n",
    "    chem_start, chem_end = c.chemical.get_word_start(), c.chemical.get_word_end()\n",
    "    dis_start, dis_end = c.disease.get_word_start(), c.disease.get_word_end()\n",
    "    if dis_start < chem_start:\n",
    "        dist = chem_start - dis_end\n",
    "    else:\n",
    "        dist = dis_start - chem_end\n",
    "    # Try to find chemical closer than @dist/2 in either direction\n",
    "    sent = c.get_parent()\n",
    "    closest_other_chem = float('inf')\n",
    "    for i in range(dis_end, min(len(sent.words), dis_end + dist / 2)):\n",
    "        et, cid = sent.entity_types[i], sent.entity_cids[i]\n",
    "        if et == 'Chemical' and cid != sent.entity_cids[chem_start]:\n",
    "            return -1\n",
    "    for i in range(max(0, dis_start - dist / 2), dis_start):\n",
    "        et, cid = sent.entity_types[i], sent.entity_cids[i]\n",
    "        if et == 'Chemical' and cid != sent.entity_cids[chem_start]:\n",
    "            return -1\n",
    "    return 0\n",
    "\n",
    "def LF_closer_dis(c):\n",
    "    # Get distance between chemical and disease\n",
    "    chem_start, chem_end = c.chemical.get_word_start(), c.chemical.get_word_end()\n",
    "    dis_start, dis_end = c.disease.get_word_start(), c.disease.get_word_end()\n",
    "    if dis_start < chem_start:\n",
    "        dist = chem_start - dis_end\n",
    "    else:\n",
    "        dist = dis_start - chem_end\n",
    "    # Try to find chemical disease than @dist/8 in either direction\n",
    "    sent = c.get_parent()\n",
    "    for i in range(chem_end, min(len(sent.words), chem_end + dist / 8)):\n",
    "        et, cid = sent.entity_types[i], sent.entity_cids[i]\n",
    "        if et == 'Disease' and cid != sent.entity_cids[dis_start]:\n",
    "            return -1\n",
    "    for i in range(max(0, chem_start - dist / 8), chem_start):\n",
    "        et, cid = sent.entity_types[i], sent.entity_cids[i]\n",
    "        if et == 'Disease' and cid != sent.entity_cids[dis_start]:\n",
    "            return -1\n",
    "    return 0"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Collects all LFs in a list:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "LFs = [\n",
    "    LF_c_cause_d, LF_c_d, LF_c_induced_d, LF_c_treat_d, LF_c_treat_d_wide,\n",
    "    LF_closer_chem, LF_closer_dis, LF_ctd_marker_c_d, LF_ctd_marker_induce,\n",
    "    LF_ctd_therapy_treat, LF_ctd_unspecified_treat, LF_ctd_unspecified_induce, LF_d_following_c,\n",
    "    LF_d_induced_by_c, LF_d_induced_by_c_tight, LF_d_treat_c, LF_develop_d_following_c, LF_far_c_d,\n",
    "    LF_far_d_c, LF_improve_before_disease, LF_in_ctd_therapy, LF_in_ctd_marker,\n",
    "    LF_in_patient_with, LF_induce, LF_induce_name, LF_induced_other,\n",
    "    LF_level, LF_measure, LF_neg_d, LF_risk_d,\n",
    "    LF_treat_d, LF_uncertain, LF_weak_assertions,\n",
    "]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Executing Labeling Functions on Spark\n",
    "\n",
    "Finally, we create a `SparkLabelAnnotator` to distribute the data, execute the labeling functions, and collect the results back into the local Snorkel database."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from snorkel.contrib.snark import SparkLabelAnnotator\n",
    "\n",
    "annotator = SparkLabelAnnotator(snorkel_session, spark_session, ChemicalDisease)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%time L_train = annotator.apply(LFs, split=0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "L_train.lf_stats(snorkel_session)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Training the Generative Model\n",
    "\n",
    "Now that the labeling functions have been executed, you can resume the chemical-disease task at Part IV in [notebook 2](https://github.com/HazyResearch/snorkel/blob/master/tutorials/cdr/CDR_Tutorial_2.ipynb)!\n",
    "\n",
    "This general approach to distributing labeling function execution can be integrated into many different Snorkel applications."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}