<jupyter_start><jupyter_text>![JohnSnowLabs](https://nlp.johnsnowlabs.com/assets/images/logo.png) [![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/JohnSnowLabs/spark-nlp-workshop/blob/master/tutorials/Certification_Trainings/Healthcare/1.4.Biomedical_NER_SparkNLP_paper_reproduce.ipynb)  Biomedical Named Entity Recognition (NER) at Scaleofficial paper : https://arxiv.org/abs/2011.06315NER in Spark NLP Colab notebooks: https://colab.research.google.com/github/JohnSnowLabs/spark-nlp-workshop/blob/master/tutorials/Certification_Trainings/Public/4.NERDL_Training.ipynb  Blogposts and videos:https://towardsdatascience.com/named-entity-recognition-ner-with-bert-in-spark-nlp-874df20d1d77NerDL worksghop (90 min): https://www.youtube.com/watch?v=YM-e4eOiQ34https://medium.com/spark-nlp/named-entity-recognition-for-healthcare-with-sparknlp-nerdl-and-nercrf-a7751b6ad571https://medium.com/atlas-research/ner-for-clinical-text-7c73caddd180<jupyter_code>import sparknlp

spark = sparknlp.start() # for GPU training >> sparknlp.start(gpu = True)

from sparknlp.base import *
from sparknlp.annotator import *

print("Spark NLP version", sparknlp.version())

print("Apache Spark version:", spark.version)

spark<jupyter_output>Spark NLP version 3.1.1
Apache Spark version: 3.0.3<jupyter_text>CoNLL Data Prep We are going to use NCBI-Disease dataset in this notebook. All the other datasets used in the official paper arw prepared using the datasets at https://github.com/cambridgeltl/MTL-Bioinformatics-2016/tree/master/data. Since the data preparation steps from official tsv dfiles to conll format we utilize here is a time consuming and proprietary process, we share only one of them (**NCBI-Disease**) as an example.<jupyter_code>!wget -q https://raw.githubusercontent.com/JohnSnowLabs/spark-nlp-workshop/master/tutorials/Certification_Trainings/Healthcare/data/NCBI_disease_official_test.conll
!wget -q https://raw.githubusercontent.com/JohnSnowLabs/spark-nlp-workshop/master/tutorials/Certification_Trainings/Healthcare/data/NCBI_disease_official_train_dev.conll
with open("NCBI_disease_official_test.conll") as f:
    train_txt =f.read()

print (train_txt[:500])
from sparknlp.training import CoNLL

training_data = CoNLL().readDataset(spark, 'NCBI_disease_official_train_dev.conll')

training_data.show(3)
training_data.count()
import pyspark.sql.functions as F

training_data.select(F.explode(F.arrays_zip('token.result', 'pos.result',  'label.result')).alias("cols")) \
.select(F.expr("cols['0']").alias("token"),
        F.expr("cols['1']").alias("pos"),
        F.expr("cols['2']").alias("ner_label")).show(truncate=50)
training_data.select(F.explode(F.arrays_zip('token.result','label.result')).alias("cols")) \
.select(F.expr("cols['0']").alias("token"),
        F.expr("cols['1']").alias("ground_truth")).groupBy('ground_truth').count().orderBy('count', ascending=False).show(100,truncate=False)

test_data = CoNLL().readDataset(spark, 'NCBI_disease_official_test.conll')

test_data.select(F.explode(F.arrays_zip('token.result','label.result')).alias("cols")) \
.select(F.expr("cols['0']").alias("token"),
        F.expr("cols['1']").alias("ground_truth")).groupBy('ground_truth').count().orderBy('count', ascending=False).show(100,truncate=False)
# You can use any word embeddings you want (Glove, Elmo, Bert, custom etc.)

glove_embeddings = WordEmbeddingsModel.pretrained('glove_6B_300','xx')\
          .setInputCols(["sentence", "token"])\
          .setOutputCol("embeddings")<jupyter_output>glove_6B_300 download started this may take some time.
Approximate size to download 426.2 MB
[OK!]<jupyter_text>**We will get the embeddings for test set and write to disk as a parquet to be able to get metrics after each epoch during training.**<jupyter_code>glove_embeddings.transform(test_data).write.parquet('test_data_with_embeddings.parquet')
!mkdir ner_logs
nerTagger = NerDLApproach()\
      .setInputCols(["sentence", "token", "embeddings"])\
      .setLabelColumn("label")\
      .setOutputCol("ner")\
      .setMaxEpochs(10)\
      .setLr(0.003)\
      .setBatchSize(8)\
      .setRandomSeed(0)\
      .setVerbose(1)\
      .setEvaluationLogExtended(True) \
      .setEnableOutputLogs(True)\
      .setIncludeConfidence(True)\
      .setTestDataset('test_data_with_embeddings.parquet')\
      .setOutputLogsPath('ner_logs') # if not set, logs will be written to ~/annotator_logs
 #    .setGraphFolder('graphs') >> put your graph file (pb) under this folder if you are using a custom graph generated thru 4.1 NerDL-Graph.ipynb notebook
 #    .setEnableMemoryOptimizer() >> if you have a limited memory and a large conll file, you can set this True to train batch by batch 
    
# do hyperparameter by tuning the params above (max epoch, LR, dropout etc.) to get better results
ner_pipeline = Pipeline(stages=[
      glove_embeddings,
      nerTagger
 ])<jupyter_output><empty_output><jupyter_text>Fitting<jupyter_code>%%time

ner_model = ner_pipeline.fit(training_data)

# 1 epoch takes around 3 min with batch size=8
# if you get an error for incompatible TF graph, use 4.1 NerDL-Graph.ipynb notebook to create a graph
#! cd ~/annotator_logs && ls -lt
!cd ner_logs && ls -lt
!tail -n 10 ner_logs/NerDLApproach_af2879065348.log<jupyter_output>Epoch 10/10 - 185.40s - loss: 421.31494 - batches: 795
Quality on test dataset: 
time to finish evaluation: 5.56s
label	 tp	 fp	 fn	 prec	 rec	 f1
I-Disease	 925	 153	 162	 0.8580705	 0.850966	 0.85450345
B-Disease	 812	 149	 148	 0.8449532	 0.84583336	 0.84539306
tp: 1737 fp: 302 fn: 310 labels: 2
Macro-average	 prec: 0.85151184, rec: 0.84839964, f1: 0.84995294
Micro-average	 prec: 0.8518882, rec: 0.84855884, f1: 0.85022026<jupyter_text>Test set evaluation<jupyter_code>import pyspark.sql.functions as F

predictions = ner_model.transform(test_data)

predictions.select(F.explode(F.arrays_zip('token.result','label.result','ner.result')).alias("cols")) \
.select(F.expr("cols['0']").alias("token"),
        F.expr("cols['1']").alias("ground_truth"),
        F.expr("cols['2']").alias("prediction")).show(truncate=False)<jupyter_output>+--------------+------------+----------+
|token         |ground_truth|prediction|
+--------------+------------+----------+
|Clustering    |O           |O         |
|of            |O           |O         |
|missense      |O           |O         |
|mutations     |O           |O         |
|in            |O           |O         |
|the           |O           |O         |
|ataxia        |B-Disease   |B-Disease |
|-             |I-Disease   |I-Disease |
|telangiectasia|I-Disease   |I-Disease |
|gene          |O           |O         |
|in            |O           |O         |
|a             |O           |O         |
|sporadic      |B-Disease   |B-Disease |
|T             |I-Disease   |I-Disease |
|-             |I-Disease   |I-Disease |
|cell          |I-Disease   |I-Disease |
|leukaemia     |I-Disease   |I-Disease |
|.             |O           |O         |
|Ataxia        |B-Disease   |B-Disease |
|-             |I-Disease   |I-Disease |
+--------------+------------+----------+
only showing top[...]<jupyter_text>Licensed user will have an access to internal NERDLMetrics module to do this more efficient and easily without going out of Spark. But open source users need to use sklearn.metrics or any other equivalent module to do the same.<jupyter_code>from sklearn.metrics import classification_report

preds_df = predictions.select(F.explode(F.arrays_zip('token.result','label.result','ner.result')).alias("cols")) \
.select(F.expr("cols['0']").alias("token"),
        F.expr("cols['1']").alias("ground_truth"),
        F.expr("cols['2']").alias("prediction")).toPandas()

print (classification_report(preds_df['ground_truth'], preds_df['prediction']))<jupyter_output>precision    recall  f1-score   support

   B-Disease       0.84      0.85      0.85       960
   I-Disease       0.86      0.85      0.85      1087
           O       0.99      0.99      0.99     22450

    accuracy                           0.98     24497
   macro avg       0.90      0.90      0.90     24497
weighted avg       0.98      0.98      0.98     24497<jupyter_text>Entity level evaluation (strict eval)<jupyter_code>!wget  -q https://raw.githubusercontent.com/JohnSnowLabs/spark-nlp-workshop/master/tutorials/Certification_Trainings/Public/utils/conll_eval.py
import conll_eval

metrics = conll_eval.evaluate(preds_df['ground_truth'].values, preds_df['prediction'].values)
# micro, macro, avg
metrics[0]
import pandas as pd
pd.DataFrame(metrics[1], columns=['entity','precision','recall','f1','support'])<jupyter_output><empty_output><jupyter_text>Ner log parser<jupyter_code>!wget -q https://raw.githubusercontent.com/JohnSnowLabs/spark-nlp-workshop/master/tutorials/Certification_Trainings/Public/utils/ner_log_parser.py
import ner_log_parser
%matplotlib inline
ner_log_parser.get_charts('/content/ner_logs/NerDLApproach_af2879065348.log')
ner_log_parser.loss_plot('/content/ner_logs/NerDLApproach_af2879065348.log')<jupyter_output><empty_output><jupyter_text>Saving the trained model<jupyter_code>ner_model.stages
ner_model.stages[1].write().overwrite().save('NER_glove_e10_b8')
!ls -lt<jupyter_output>total 217684
drwxr-xr-x  4 root root      4096 Jun 24 11:36 NER_glove_e10_b8
drwxr-xr-x  2 root root      4096 Jun 24 11:35 __pycache__
-rw-r--r--  1 root root      3826 Jun 24 11:35 ner_log_parser.py
-rw-r--r--  1 root root      7431 Jun 24 11:34 conll_eval.py
drwxr-xr-x  2 root root      4096 Jun 24 10:58 ner_logs
drwxr-xr-x  2 root root      4096 Jun 24 10:57 test_data_with_embeddings.parquet
-rw-r--r--  1 root root   2136193 Jun 24 10:55 NCBI_disease_official_train_dev.conll
-rw-r--r--  1 root root    328172 Jun 24 10:55 NCBI_disease_official_test.conll
-rw-r--r--  1 root root 220400553 Jun 17 05:28 spark-3.0.3-bin-hadoop2.7.tgz
drwxr-xr-x 13 1000 1000      4096 Jun 17 05:27 spark-3.0.3-bin-hadoop2.7
drwxr-xr-x  1 root root      4096 Jun 15 13:37 sample_data<jupyter_text>Prediction Pipeline<jupyter_code>document = DocumentAssembler()\
        .setInputCol("text")\
        .setOutputCol("document")

sentence = SentenceDetector()\
        .setInputCols(['document'])\
        .setOutputCol('sentence')

token = Tokenizer()\
        .setInputCols(['sentence'])\
        .setOutputCol('token')

glove_embeddings = WordEmbeddingsModel.pretrained('glove_6B_300','xx')\
        .setInputCols(["sentence", "token"])\
        .setOutputCol("embeddings")
  
loaded_ner_model = NerDLModel.load("NER_glove_e10_b8")\
        .setInputCols(["sentence", "token", "embeddings"])\
        .setOutputCol("ner")

converter = NerConverter()\
        .setInputCols(["sentence", "token", "ner"])\
        .setOutputCol("ner_span")

ner_prediction_pipeline = Pipeline(stages = [
        document,
        sentence,
        token,
        glove_embeddings,
        loaded_ner_model,
        converter
])
empty_data = spark.createDataFrame([['']]).toDF("text")

prediction_model = ner_prediction_pipeline.fit(empty_data)

text = '''
A 28-year-old female with a history of gestational diabetes mellitus diagnosed eight years prior to presentation and subsequent type two diabetes mellitus ( T2DM ), one prior episode of HTG-induced pancreatitis three years prior to presentation , associated with an acute hepatitis , and obesity with a body mass index ( BMI ) of 33.5 kg/m2 , presented with a one-week history of polyuria , polydipsia , poor appetite , and vomiting .
'''

sample_data = spark.createDataFrame([[text]]).toDF("text")

sample_data.show(truncate=100)
preds = prediction_model.transform(sample_data)

preds.select(F.explode(F.arrays_zip("ner_span.result","ner_span.metadata")).alias("entities")) \
.select(F.expr("entities['0']").alias("chunk"),
        F.expr("entities['1'].entity").alias("entity")).show(truncate=False)
from sparknlp.base import LightPipeline

light_model = LightPipeline(prediction_model)
result = light_model.annotate(text)

list(zip(result['token'], result['ner']))
import pandas as pd

result = light_model.fullAnnotate(text)

ner_df= pd.DataFrame([(int(x.metadata['sentence']), x.result, x.begin, x.end, y.result) for x,y in zip(result[0]["token"], result[0]["ner"])], 
                      columns=['sent_id','token','start','end','ner'])
ner_df.head(20)<jupyter_output><empty_output><jupyter_text>Highlight Entities<jupyter_code>ann_text = light_model.fullAnnotate(text)[0]
ann_text.keys()
from sparknlp_display import NerVisualizer

visualiser = NerVisualizer()
print ('Standard Output')
visualiser.display(ann_text, label_col='ner_span', document_col='document')<jupyter_output>Standard Output
{'fraction_non_alphanumeric': 0.1418216199180586, 'fraction_numerical': 0.04396470217459817, 'mean_word_length': 3.7880045265937383, 'pattern_counts': {'":': 0, '<': 38, '<?xml version=': 0, '>': 44, 'https://': 14, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 5, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '11', 'dir': 'jupyter-structured-clean-dedup', 'id': '47369', 'license': 'Apache-2.0', 'n_tokens_mistral': 4844, 'n_tokens_neox': 4335, 'n_words': 853, 'path': 'jupyter/docker_enterprise/docker_notebooks/Spark_NLP/Healthcare/1.4.Biomedical_NER_SparkNLP_paper_reproduce.ipynb', 'repo_name': 'fcivardi/spark-nlp-workshop', 'size': '12692'}