ElasticSearchBulkNoteExporter.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Extract note text for a large set of notes from ElasticSearch\n",
    "\n",
    "This notebook script:\n",
    "1. Connects to an ElasticSearch instance.\n",
    "2. Extracts a set of elements in an index in the ES instance. The index is based on text extracted from notes in Neptune.\n",
    "3. Distributes the extracted elements in files in a particular folder structure on a Linux \"local machine\".\n",
    "\n",
    "## Background\n",
    "The extraction processing is brittle, and subject to many interruptions. A Jupyter notebook will not be able to extract more than around 25K notes from ES in the current configuration before something interrupts. The most likely form of interruption is a VPN timeout.\n",
    "\n",
    "It is necessary to extract in well-defined batches so that we can keep track of progress.\n",
    "\n",
    "## Output file structure\n",
    "Output files will be distributed as follows:\n",
    "1. Folder structure\n",
    "- There will be subfolders of the output root corresponding to the numbers 0-9.\n",
    "- Under each numbered folder will be subfolders for all study ids that end in the number.\n",
    "- In each study id folder will be text files for all notes extracted for the study id.\n",
    "2. Each file will be named with the STUDY_NOTE_CSN_ID.\n",
    "\n",
    "For example, the note with STUDY_NOTE_CSN_ID 25 for patient with study ID 998 will be in a file named 25.txt in folder path _output root_/8/998.\n",
    "\n",
    "\n",
    "\n",
    "## Assumptions\n",
    "1. An active VPN channel has been established . (This script was intended to run in DBMI's Jupyter environment, so a user cannot start a Jupyter session without first\n",
    "establishing a VPN connection.)\n",
    "2. You have an account with access to the Elastic Search instance.\n",
    "3. File writes have been tested only to the server that hosts the Jupyter environment. Ideally, the script would write to a Globus collection.\n",
    "4. An extremely large number of notes (millions).\n",
    "5. The notes to be extracted are specified in a spreadsheet with at least the following columns:\n",
    "- STUDY_ID\n",
    "- NOTE_CSN_ID\n",
    "- STUDY_NOTE_ID\n",
    "- STUDY_NOTE_CSN_ID\n",
    "\n",
    "\n",
    "## To extract note text\n",
    "1. In the cell marked **LOCAL ELASTICSEARCH CONNECTION**, follow the instructions to set up a local configuration file for connection to the ElasticSearch instance. You should not need to modify the code in the section.\n",
    "2. Move to the cell marked **GENERATION PARAMETERS**.\n",
    "\n",
    "- Edit values of variables that the script uses to extract a set of notes.  This includes the \"ordinal range\" variables **note_start** and **note_end**. For example, **note_start** = 25000 and **note_end** = 50000 will extract the 25000th through the 50000th notes in the input CSV.\n",
    "\n",
    "- From the Run menu, Run All Above Selected cell.\n",
    "3. In the cell marked **RUN SCRIPT**, run the script. \n",
    "\n",
    "The script will:\n",
    "- write extracted output to the output folder path\n",
    "- write log files for both all files and missing files\n",
    "\n",
    "\n",
    "## To summarize work:\n",
    "After running all batches, you can use the code in the **CONSOLIDATE LIST FOR ALL BATCHES** cell.\n",
    "\n",
    "\n",
    "## Utility code\n",
    "The code blocks identified as **UTILITY CODE*** contain utility functions for working with ElasticSearch. They should not need to be edited. Ideally, this utility code would be encapsulated in a Python class and imported.\n",
    "\n",
    "\n",
    "## Logging\n",
    "It is necessary to work in batches (e.g., 100K notes at a time) to avoid interruptions. It is also likely that some of the requested notes will not be in the ElasticSearch index. To keep track of both what *was* extracted and what *was not* extracted, the script will log progress for a batch in a spreadsheet. \n",
    "\n",
    "The spreadsheet will be named with format **NOTES-STATUS-_start_-_end_.csv**, where _start_ and _end_ correspond to a ordinal number of notes in the input spreadsheet. For example, a file named NOTES-STATUS-10000-20000.csv will track the status of extraction of the 10000th through the 20000th note.\n",
    "\n",
    "The notebook will also build a file named **MISSING-_start_-_end_.csv**, containing only the values of STUDY_NOTE_CSN_ID for notes that were not extracted.\n",
    "\n",
    "Finally, the script logs uses Python logging to document information on the overall run.\n",
    "\n",
    "## Parallelization\n",
    "If you need to extract a very large number of notes, you can reduce extraction time by establishing multiple parallel streams. To do this:\n",
    "1. Create multiple copies of this script.\n",
    "2. In each copy of the script, edit the values of **logfile** and **logapp** in the **Python logging** cell. This will allow you to track progress of each instance of the script in separate logs.\n",
    "3. Run each instance with a different ordinal range. For example, to extract 100,000 notes in 4 batches of 25K, use:\n",
    "- 0 to 25000\n",
    "- 25001 to 50000\n",
    "- 50001 to 75000\n",
    "- 75001 to 100000\n",
    "\n",
    "Note that this form of parallelization requires discipline to keep track of what you've done, especially when working with a large number of files.\n",
    "Fortunately, the actual extraction of a single note does not take long. In the worst case, just re-extract a batch."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# UTILITY CODE\n",
    "\n",
    "import configparser\n",
    "import re\n",
    "import csv\n",
    "from pathlib import Path\n",
    "\n",
    "import pandas as pd\n",
    "import os\n",
    "\n",
    "# Elasticsearch python libs\n",
    "from elasticsearch import Elasticsearch\n",
    "from elasticsearch import helpers\n",
    "from ssl import create_default_context\n",
    "from elasticsearch_dsl import Search\n",
    "\n",
    "import time\n",
    "from datetime import timedelta\n",
    "from datetime import datetime\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# UTILITY CODE\n",
    "\n",
    "####################################################\n",
    "# Helper functions, add any custom functions here\n",
    "####################################################\n",
    "\n",
    "# function make substitutions for newlines\n",
    "def reformat_text(text):\n",
    "    #return re.sub(r'[^\\s]([ ]{5,})[^\\s]', '<br>', text)\n",
    "    return re.sub(r'[ ]{3}[ ]+', '\\n', text)\n",
    "\n",
    "\n",
    "def space_checker(text):\n",
    "    SP_THRESHOLD = 6\n",
    "    text_pos = 0\n",
    "    sp_pos = 0\n",
    "    for c in text:\n",
    "        if c.isspace():\n",
    "            if sp_cnt == 0:\n",
    "                sp_pos = text_pos\n",
    "            sp_cnt += 1\n",
    "            if sp_cnt > SP_THRESHOLD:\n",
    "                print('found space groups of threshold @ {},{}'.format(sp_pos, text_pos))\n",
    "        else:\n",
    "            sp_pos = 0\n",
    "            sp_cnt = 0\n",
    "        text_pos += 1\n",
    "            \n",
    "def get_single_report(id):\n",
    "    reports = []\n",
    "    s = Search(using=es, index=LOCAL['elasticsearch']['INDEX']) \\\n",
    "            .query(\"match\", NOTE_CSN_ID=id)\n",
    "    \n",
    "    # convert to a dictionary for the DSL\n",
    "    body = s.to_dict()\n",
    "    \n",
    "     # do the search with the body\n",
    "    response = es.search(body=body)\n",
    "    \n",
    "    hits = response['hits']['hits']   \n",
    "    text = reformat_text(hits[0]['_source']['NOTE_TEXT'])   # experimental code\n",
    "    #text = hits[0]['_source']['NOTE_TEXT']\n",
    "    #space_checker(hits[0]['_source']['NOTE_TEXT'])     # experimental code\n",
    "\n",
    "    return text\n",
    "\n",
    "# parameters:  \n",
    "#    fname, report text,  html flag (true/false)\n",
    "def save_to_file(fname, text, outdir, html):\n",
    "    \n",
    "    if html:\n",
    "        filename = '{}/{}.html'.format(outdir, fname)\n",
    "    else:\n",
    "        filename = '{}/{}.txt'.format(outdir, fname)\n",
    "        \n",
    "    with open(filename, 'w') as f:\n",
    "        if html:\n",
    "            f.write('<html>\\n')\n",
    "            f.write('<p>\\n')\n",
    "            \n",
    "        # write the report text\n",
    "        f.write(text)\n",
    "        \n",
    "        if html:\n",
    "            f.write('</p>\\n')\n",
    "            f.write('</html>\\n')\n",
    "    f.close()\n",
    "    \n",
    "def get_file_size(fname, text, outdir, html):\n",
    "    if html:\n",
    "        filename = '{}/{}.html'.format(outdir, fname)\n",
    "    else:\n",
    "        filename = '{}/{}.txt'.format(outdir, fname)\n",
    "    \n",
    "    return os.path.getsize(filename)\n",
    "    \n",
    "       "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# UTILITY CODE\n",
    "\n",
    "##############################################################\n",
    "# Load Elasticsearch settings from a common config file stored in the local file system of the R3 Jupyter host machine.\n",
    "# Do not modify\n",
    "###############################################################\n",
    "config = configparser.ConfigParser()\n",
    "config.read('/data/R3_Library/es_settings.ini')\n",
    "COMMON_SETTINGS = config\n",
    "CERT = COMMON_SETTINGS['cert']['CERT_FILE']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# LOCAL ELASTICSEARCH CONFIGURATION\n",
    "\n",
    "# Create a local Kibana (ElasticSearch) configuration file with the following content:\n",
    "\n",
    "#-----------content\n",
    "\n",
    "#[cert]\n",
    "#CERT_FILE=<path to the common certificate--e.g., '/data/R3_Library/Certs/notes01.revchain.crt'>\n",
    "\n",
    "#[elasticsearch]\n",
    "#INDEX=<custom index name or use 'neptune'>\n",
    "#HOST=<host name for the Kibana instance--e.g., 'dbmi-neptune-notes-01.dbmi.pitt.edu'>\n",
    "#PORT=<port>\n",
    "\n",
    "#[user]\n",
    "#USERNAME=<your username for the Kibana instance>\n",
    "#PASSWORD=<your password for the Kibana instance>\n",
    "\n",
    "#-----------content\n",
    "\n",
    "# NOTE:  if you are using your own index you can specify that name in INDEX;  if you have\n",
    "#        have access to the full index use 'neptune'\n",
    "\n",
    "#Indicate the path to the file. For example, you can store the file in the same Jupyter directory in which you run this notebook.\n",
    "local_config_path = 'es_local_settings.ini'\n",
    "\n",
    "config.read(local_config_path)\n",
    "LOCAL = config\n",
    "\n",
    "# set the ssl context to communicate via https\n",
    "ssl_context = create_default_context(cafile=CERT)\n",
    "\n",
    "# initalize the Elasticsearch client\n",
    "es = Elasticsearch(COMMON_SETTINGS['elasticsearch']['HOST'],\n",
    "\t\thttp_auth=(LOCAL['user']['USERNAME'], LOCAL['user']['PASSWORD']),\n",
    "\t\tscheme=\"https\",\n",
    "\t\tport=COMMON_SETTINGS['elasticsearch']['PORT'],\n",
    "\t\tssl_context=ssl_context\n",
    "\t\t)\n",
    "\n",
    "# check for activate connection\n",
    "if not es.ping():\n",
    "    raise ValueError(\"Connection failed\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#UTILITY CODE\n",
    "#Python logging.\n",
    "\n",
    "#Assumptions:\n",
    "#1. Log file will be written to application folder.\n",
    "#2. The application folder contains the ini file for logging.\n",
    "#3. The ini file content conforms to recommendations in https://www.datadoghq.com/blog/python-logging-best-practices/\n",
    "\n",
    "import logging.config\n",
    "log_dir = '' # Application folder\n",
    "log_config = 'logging.ini' # Configures logging\n",
    "logfile = 'NoteExporter3.log'\n",
    "logapp = 'NoteExporter3'\n",
    "logger = logging.getLogger(logapp)\n",
    "logging.config.fileConfig(log_config, disable_existing_loggers=False, defaults={'log_file':logfile})\n",
    "\n",
    "#Filter out elasticsearch info.\n",
    "logger_es = logging.getLogger('elasticsearch')\n",
    "logger_es.setLevel(logging.WARN)\n",
    "\n",
    "def print_and_logger_info(message: str) -> None:\n",
    "    print(message)\n",
    "    logger.info(message)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#GENERATION PARAMETERS\n",
    "#--------\n",
    "#1. Full path to file of notes to extract.\n",
    "id_file = '/data/R3_Library/2887_Hochheiser_Melanoma_Encounter/input/R3_2887_HOCHHEISER_ENCOUNTER_NOTES_2022_07_07.csv'\n",
    "\n",
    "#2. Full paths to output directories\n",
    "outdir = '/data/R3_Library/2887_Hochheiser_Melanoma_Encounter/output'\n",
    "trackingdir = outdir + '/tracking'\n",
    "missingdir = outdir + '/missing'\n",
    "jupytertrackingdir = os.getcwd()+'/tracking'\n",
    "jupytermissingdir = os.getcwd()+'/missing'\n",
    "\n",
    "#3. Ordinal range of notes to extract in this batch.\n",
    "note_start = 2450000\n",
    "note_end=2475000\n",
    "\n",
    "#Set read_input to False if doing repeated runs, so that the script will use the input already in memory.\n",
    "read_input = True\n",
    "#--------"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# RUN SCRIPT\n",
    "\n",
    "print_and_logger_info('************************')\n",
    "print_and_logger_info('NOTE EXTRACTION STARTED')\n",
    "\n",
    "#Output tracking files.\n",
    "tracking_col_names =  ['STUDY_ID','STUDY_NOTE_CSN_ID','status', 'size','date','time']\n",
    "\n",
    "dfTracking = pd.DataFrame(columns = tracking_col_names)\n",
    "tracking_file = 'NOTES_STATUS_' + str(note_start) + '_' + str(note_end) +'.csv'\n",
    "tracking_path = trackingdir + '/' + tracking_file\n",
    "\n",
    "missing_col_names = ['STUDY_NOTE_CSN_ID']\n",
    "dfmissing = pd.DataFrame(columns = missing_col_names)\n",
    "missing_file = 'MISSING_' + str(note_start) + '_' + str(note_end) +'.csv'\n",
    "missing_path = missingdir + '/' + missing_file\n",
    "\n",
    "if read_input:\n",
    "    print_and_logger_info(f'Reading input spreadsheet: {id_file}')\n",
    "    csvreader = pd.read_csv(id_file, sep=',', dtype='str', na_filter=False)[['STUDY_ID','NOTE_CSN_ID','STUDY_NOTE_CSN_ID']]\n",
    "else:\n",
    "    print_and_logger_info(f'Using input spreadsheet data in memory for file: {id_file}')\n",
    "\n",
    "print_and_logger_info(f'Total number of notes in spreadsheet: {len(csvreader.index)}')\n",
    "print_and_logger_info(f'Range of notes to extract: {str(note_start)} to {str(note_end)}')\n",
    "print_and_logger_info('Extracting notes')\n",
    "\n",
    "#Track overall generation time.\n",
    "start_time = time.time()\n",
    "\n",
    "\n",
    "# Because of the large number of files, files will be distributed to the following output file folder structure:\n",
    "# output\n",
    "# --number from 0-9, corresponding to the last number of the study ID.\n",
    "# |\n",
    "#   --study ID\n",
    "#   |\n",
    "#     --study_note_csn_id.txt\n",
    "\n",
    "#Create the output subfolders.\n",
    "os.system(f\"mkdir -p {outdir}\")\n",
    "for i in range(0,10):\n",
    "    numdir = outdir+\"/\"+str(i)\n",
    "    os.system(f\"mkdir -p {numdir}\")\n",
    "    \n",
    "#Create the tracking folders.\n",
    "os.system(f'mkdir -p {trackingdir}')\n",
    "os.system(f'mkdir -p {missingdir}')\n",
    "    \n",
    "percentdisplay = 0\n",
    "for index, rows in csvreader.iterrows():\n",
    "    \n",
    "    if index > note_end:\n",
    "        break\n",
    "    if index >= note_start and index <= note_end:\n",
    "        #Poor man's TQDM.\n",
    "        percentdone = int(round(1 - (note_end-index)/(note_end-note_start),1)*100)\n",
    "        if percentdone % 10 == 0 and percentdone <100:\n",
    "            if percentdone != percentdisplay:\n",
    "                percentdisplay = percentdone\n",
    "                print(f'{percentdone}% complete')\n",
    "        \n",
    "        try:\n",
    "            #Extract note text from Elastic Search, using the Epic NOTE_CSN_ID.\n",
    "            text = get_single_report(rows['NOTE_CSN_ID'])\n",
    "            \n",
    "            #Create the output subfolder for the study id.\n",
    "            thepath = outdir+\"/\"+rows['STUDY_ID'][-1]+\"/\"+rows['STUDY_ID']\n",
    "            os.system(f\"mkdir -p {thepath}\")\n",
    "            \n",
    "            #Write the extracted text.\n",
    "            save_to_file(rows['STUDY_NOTE_CSN_ID'], text, thepath, False)\n",
    "            filesize = get_file_size(rows['STUDY_NOTE_CSN_ID'], text, thepath, False)\n",
    "            \n",
    "            #Indicate the status.\n",
    "            genstatus = 'extracted'\n",
    "            \n",
    "        except:\n",
    "            #Indicate the status.\n",
    "            genstatus = 'failed'\n",
    "            filesize = 0\n",
    "            \n",
    "            #Log the missing file.\n",
    "            dfmissing.loc[len(dfmissing.index)] = [rows['STUDY_NOTE_CSN_ID']]\n",
    "    \n",
    "        #Log the status.\n",
    "        dictStatus={'STUDY_ID':[rows['STUDY_ID']],'STUDY_NOTE_CSN_ID':[rows['STUDY_NOTE_CSN_ID']],'status':[genstatus],'size':[filesize],'date':[datetime.now().strftime(\"%m/%d/%Y\")],'time':[datetime.now().strftime(\"%H:%M:%S\")]}\n",
    "        dfStatus = pd.DataFrame(data=dictStatus,columns=tracking_col_names)\n",
    "        dfTracking = pd.concat([dfTracking,dfStatus])\n",
    "\n",
    "    \n",
    "\n",
    "#Write logs to two locations:\n",
    "#1. In the output path (presumably in R3 data path)\n",
    "#2. Locally to Jupyter (so it is not necessary to copy in order to view it easily.)\n",
    "\n",
    "dfTracking.to_csv(tracking_path, mode='w', header=True, index=False)\n",
    "dfTracking.to_csv(os.getcwd()+ tracking_file, mode='w', header=True, index=False)\n",
    "dfmissing.to_csv(missing_path, mode='w', header=True, index=False)\n",
    "dfmissing.to_csv(os.getcwd()+ missing_file, mode='w', header=True, index=False)\n",
    "\n",
    "#Log statistics for batch.\n",
    "print_and_logger_info(f'Completed batch run for notes in range: {str(note_start)} to {str(note_end)}')\n",
    "print_and_logger_info(f'Number of missing notes in this batch: {len(dfmissing.index)}')\n",
    "elapsed_time = time.time() - start_time\n",
    "print_and_logger_info(f'Extraction time for this batch: {\"{:0>8}\".format(str(timedelta(seconds=elapsed_time)))}')\n",
    "totsize = dfTracking['size'].sum()\n",
    "print_and_logger_info(f'Total size of files extracted for this batch: {totsize/1024:.2f} KB ({totsize/1024**2:.3f} MB)')\n",
    "print_and_logger_info('DONE')\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Consolidate lists\n",
    "\n",
    "Union all of the NOTES_STATUS..csv and  MISSING..csv files into single files."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def consolidate_list(listtype):\n",
    "    \n",
    "    tracking_col_names =  ['STUDY_ID','STUDY_NOTE_CSN_ID','status', 'size','date','time']\n",
    "    missing_col_names = ['STUDY_NOTE_CSN_ID']\n",
    "\n",
    "    if listtype =='missing':\n",
    "        cols = missing_col_names\n",
    "        listdir = missingdir\n",
    "        jupyterlistdir = jupytermissingdir\n",
    "        allfile = 'MISSING_ALL.csv'\n",
    "        \n",
    "    elif listtype == 'status':\n",
    "        cols = tracking_col_names\n",
    "        listdir = trackingdir\n",
    "        jupyterlistdir = jupytertrackingdir\n",
    "        allfile = 'NOTES_STATUS_ALL.csv'\n",
    "    else:\n",
    "        raise('Unknown list type: ',listtype)\n",
    "    \n",
    "    allpath = os.path.join(listdir,allfile)\n",
    "    dfall = pd.DataFrame(columns = cols)\n",
    "\n",
    "    #Get list of files in the missing directory.\n",
    "    listfiles = os.listdir(listdir)\n",
    "\n",
    "    #Append contents of each file to the consolidated file.\n",
    "    for path in os.listdir(listdir):\n",
    "        if path !=allfile:\n",
    "            path_file = os.path.join(listdir,path)\n",
    "            dflist = pd.read_csv(path_file, sep=',', dtype='str', na_filter=False)\n",
    "            dfall = pd.concat([dfall,dflist],ignore_index=True).drop_duplicates()\n",
    "\n",
    "    #Write consolidated file.\n",
    "    dfall.to_csv(allpath, mode='w', header=True, index=False)\n",
    "    dfall.to_csv(os.path.join(jupyterlistdir,allfile), mode='w', header=True, index=False)\n",
    "    \n",
    "    return\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# CONSOLIDATE LISTS FOR ALL BATCHES\n",
    "\n",
    "# argument: \n",
    "#   missing - generate a 'MISSING_ALL.csv' file\n",
    "#   status - generate a 'NOTES_STATUS_ALL.csv' file\n",
    "consolidate_list('missing')"
   ]
  }
 ],
 "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.8.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}