<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/23.Drug_Normalizer.ipynb)  23.Clinical Drug Normalizer  New Annotator that transforms text to the format used in the RxNorm and SNOMED standards It takes in input annotated documents of type Array\[AnnotatorType\](DOCUMENT) and gives as output annotated document of type AnnotatorType.DOCUMENT .Parameters are:- inputCol: input column name string which targets a column of type Array(AnnotatorType.DOCUMENT).- outputCol: output column name string which targets a column of type AnnotatorType.DOCUMENT.- lowercase: whether to convert strings to lowercase. Default is False.- policy: rule to remove patterns from text.  Valid policy values are:    + **"all"**,     + **"abbreviations"**,     + **"dosages"**    Defaults is "all". "abbreviation" policy used to expend common drugs abbreviations, "dosages" policy used to convert drugs dosages and values to the standard form (see examples bellow).  Examples of transformation:    1) "Sodium Chloride/Potassium Chloride 13bag"  >>>  "Sodium Chloride / Potassium Chloride **13 bag**" : add extra spaces in the form entity2) "interferon alfa-2b 10 million unit ( 1 ml ) injec" >>> "interferon alfa - 2b 10000000 unt ( 1 ml ) injection " : convert **10 million unit** to the **10000000 unt**, replace **injec** with **injection**3) "aspirin 10 meq/ 5 ml oral sol" >>> "aspirin 2 meq/ml oral solution" : normalize **10 meq/ 5 ml** to the **2 meq/ml**, extend abbreviation **oral sol** to the **oral solution**4) "adalimumab 54.5 + 43.2 gm" >>> "adalimumab 97700 mg" : combine **54.5 + 43.2** and normalize **gm** to **mg**5) "Agnogenic one  half cup" >>> "Agnogenic 0.5 oral solution" : replace **one  half** to the **0.5**, normalize **cup** to the **oral solution**<jupyter_code>import json, os
from google.colab import files

if 'spark_jsl.json' not in os.listdir():
  license_keys = files.upload()
  os.rename(list(license_keys.keys())[0], 'spark_jsl.json')

with open('spark_jsl.json') as f:
    license_keys = json.load(f)

# Defining license key-value pairs as local variables
locals().update(license_keys)
os.environ.update(license_keys)
# Installing pyspark and spark-nlp
! pip install --upgrade -q pyspark==3.1.2 spark-nlp==$PUBLIC_VERSION

# Installing Spark NLP Healthcare
! pip install --upgrade -q spark-nlp-jsl==$JSL_VERSION  --extra-index-url https://pypi.johnsnowlabs.com/$SECRET
import json
import os

from pyspark.ml import Pipeline,PipelineModel
from pyspark.sql import SparkSession

import sparknlp_jsl
import sparknlp

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

params = {"spark.driver.memory":"16G",
          "spark.kryoserializer.buffer.max":"2000M",
          "spark.driver.maxResultSize":"2000M"}

spark = sparknlp_jsl.start(license_keys['SECRET'],params=params)

print("Spark NLP Version :", sparknlp.version())
print("Spark NLP_JSL Version :", sparknlp_jsl.version())

spark
 # if you want to start the session with custom params as in start function above

def start(secret):
    builder = SparkSession.builder \
        .appName("Spark NLP Licensed") \
        .master("local[*]") \
        .config("spark.driver.memory", "16G") \
        .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
        .config("spark.kryoserializer.buffer.max", "2000M") \
        .config("spark.jars.packages", "com.johnsnowlabs.nlp:spark-nlp_2.11:"+version)  \
        .config("spark.jars", "https://pypi.johnsnowlabs.com/"+secret+"/spark-nlp-jsl-"+jsl_version+".jar")
      
    return builder.getOrCreate()

# spark = start(secret)
# Sample data
data_to_normalize = spark.createDataFrame([
            ("A", "Sodium Chloride/Potassium Chloride 13bag", "Sodium Chloride / Potassium Chloride 13 bag"),
            ("B", "interferon alfa-2b 10 million unit ( 1 ml ) injec", "interferon alfa - 2b 10000000 unt ( 1 ml ) injection"),
            ("C", "aspirin 10 meq/ 5 ml oral sol", "aspirin 2 meq/ml oral solution")
        ]).toDF("cuid", "text", "target_normalized_text")

data_to_normalize.show(truncate=100)
# Annotator that transforms a text column from dataframe into normalized text (with all policy)

document_assembler = DocumentAssembler()\
    .setInputCol("text")\
    .setOutputCol("document")

policy = "all"

drug_normalizer = DrugNormalizer() \
    .setInputCols("document") \
    .setOutputCol("document_normalized") \
    .setPolicy(policy)

drug_normalizer_pipeline = Pipeline(stages=[
    document_assembler,
    drug_normalizer
    ])

ds = drug_normalizer_pipeline.fit(data_to_normalize).transform(data_to_normalize)

ds = ds.selectExpr("document", "target_normalized_text", "explode(document_normalized.result) as all_normalized_text")
ds.show(truncate = False)
# Annotator that transforms a text column from dataframe into normalized text (with abbreviations only policy)

policy = "abbreviations"

drug_normalizer_abb = DrugNormalizer() \
    .setInputCols("document") \
    .setOutputCol("document_normalized_abbreviations") \
    .setPolicy(policy)

ds = drug_normalizer_abb.transform(ds)

ds = ds.selectExpr("document", "target_normalized_text", "all_normalized_text", "explode(document_normalized_abbreviations.result) as abbr_normalized_text")
ds.select("target_normalized_text", "all_normalized_text", "abbr_normalized_text").show(truncate=1000)
# Transform a text column from dataframe into normalized text (with dosages only policy)

policy = "dosages"

drug_normalizer_abb = DrugNormalizer() \
    .setInputCols("document") \
    .setOutputCol("document_normalized_dosages") \
    .setPolicy(policy)

ds = drug_normalizer_abb.transform(ds)

ds.selectExpr("target_normalized_text", "all_normalized_text", "explode(document_normalized_dosages.result) as dos_normalized_text").show(truncate=1000)<jupyter_output>+----------------------------------------------------+----------------------------------------------------+------------------------------------------------+
|                              target_normalized_text|                                 all_normalized_text|                             dos_normalized_text|
+----------------------------------------------------+----------------------------------------------------+------------------------------------------------+
|         Sodium Chloride / Potassium Chloride 13 bag|         Sodium Chloride / Potassium Chloride 13 bag|     Sodium Chloride / Potassium Chloride 13 bag|
|interferon alfa - 2b 10000000 unt ( 1 ml ) injection|interferon alfa - 2b 10000000 unt ( 1 ml ) injection|interferon alfa - 2b 10000000 unt ( 1 ml ) injec|
|                      aspirin 2 meq/ml oral solution|                      aspirin 2 meq/ml oral solution|                       aspirin 2 meq/ml oral sol|
+----------------------------------------------------+----[...]<jupyter_text>Apply normalizer only on NER chunks<jupyter_code>documentAssembler = DocumentAssembler()\
    .setInputCol("text")\
    .setOutputCol("document")
 
# Sentence Detector annotator, processes various sentences per line
sentenceDetector = SentenceDetector()\
    .setInputCols(["document"])\
    .setOutputCol("sentence")
  
# Tokenizer splits words in a relevant format for NLP
tokenizer = Tokenizer()\
    .setInputCols(["sentence"])\
    .setOutputCol("token")\
    .addSplitChars(";")
  
# Clinical word embeddings trained on PubMED dataset
word_embeddings = WordEmbeddingsModel.pretrained("embeddings_clinical", "en", "clinical/models")\
    .setInputCols(["sentence", "token"])\
    .setOutputCol("embeddings")
 
# Extract entities with NER model posology
posology_ner = MedicalNerModel.pretrained("ner_posology_large", "en", "clinical/models") \
    .setInputCols(["sentence", "token", "embeddings"]) \
    .setOutputCol("ner_posology")
  
# Group extracted entities into the chunks
ner_converter = NerConverter()\
    .setInputCols(["sentence", "token", "ner_posology"])\
    .setOutputCol("ner_chunk_posology")
 
# Convert extracted entities to the doc with chunks in metadata
c2doc = Chunk2Doc()\
    .setInputCols("ner_chunk_posology")\
    .setOutputCol("chunk_doc") 

# Transform a chunk document into normalized text
drug_normalizer = DrugNormalizer() \
    .setInputCols("chunk_doc") \
    .setOutputCol("document_normalized_dosages")\
    .setPolicy("all")

nlpPipeline = Pipeline(stages=[
    documentAssembler, 
    sentenceDetector,
    tokenizer,
    word_embeddings,
    posology_ner,
    ner_converter,
    c2doc,
    drug_normalizer])

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

model = nlpPipeline.fit(empty_data)
! wget -q https://s3.amazonaws.com/auxdata.johnsnowlabs.com/public/resources/en/pubmed/pubmed_sample_text_small.csv
import pyspark.sql.functions as F

pubMedDF = spark.read\
                .option("header", "true")\
                .csv("pubmed_sample_text_small.csv")\
                
pubMedDF.show(truncate=50)
result = model.transform(pubMedDF.limit(100))
result.show(2)
import pyspark.sql.functions as F
result.select(F.explode('document_normalized_dosages.result')).show(truncate=100)<jupyter_output>+-----------------------------------------------------------------------------+
|                                                                          col|
+-----------------------------------------------------------------------------+
|G - protein - activated inwardly rectifying potassium ( GIRK ) channel family|
|                                                               anthracyclines|
|                                                                      taxanes|
|                                                                 usefulnessof|
|                                                                  vinorelbine|
|                                                                  vinorelbine|
|                                                               anthracyclines|
|                                                                      taxanes|
|                                                                 25 mg /m ( 2|
|                                       [...]
{'fraction_non_alphanumeric': 0.1588649744233182, 'fraction_numerical': 0.017083293118424863, 'mean_word_length': 2.563273727647868, 'pattern_counts': {'":': 3, '<': 7, '<?xml version=': 0, '>': 22, 'https://': 6, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 6, 'word_list_counts': {'cursed_substrings.json': 23, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '2', 'dir': 'jupyter-structured-clean-dedup', 'id': '63352', 'license': 'Apache-2.0', 'n_tokens_mistral': 3215, 'n_tokens_neox': 2946, 'n_words': 722, 'path': 'tutorials/Certification_Trainings/Healthcare/23.Drug_Normalizer.ipynb', 'repo_name': 'iamvarol/spark-nlp-workshop', 'size': '10361'}