<jupyter_start><jupyter_text>Vivekn Sentiment AnalysisIn the following example, we walk-through Sentiment Analysis training and prediction using Spark NLP Annotators.The ViveknSentimentApproach annotator will compute [Vivek Narayanan algorithm](https://arxiv.org/pdf/1305.6143.pdf) with either a column in training dataset with rows labelled 'positive' or 'negative' or a folder full of positive text and a folder with negative text. Using n-grams and negation of sequences, this statistical model can achieve high accuracy if trained properly.Spark can be leveraged in training by utilizing ReadAs.Dataset setting. Spark will be used during prediction by default.We also include in this pipeline a spell checker which shall correct our sentences for better Sentiment Analysis accuracy.  Spark `2.4` and Spark NLP `2.0.1`  1. Call necessary imports and set the resource path to read local data files<jupyter_code>#Imports
import time
import sys
import os
#sys.path.append('../../')

from pyspark.ml import Pipeline, PipelineModel
from sparknlp.annotator import *
from pyspark.sql.functions import array_contains
from sparknlp.base import DocumentAssembler, Finisher

#Setting location of resource Directory
resource_path= "../../../src/test/resources/"<jupyter_output><empty_output><jupyter_text>2. Load SparkSession if not already there<jupyter_code>spark = SparkSession.builder \
    .appName("VivekNarayanSentimentApproach")\
    .master("local[*]")\
    .config("spark.driver.memory","8G")\
    .config("spark.driver.maxResultSize", "2G")\
    .config("spark.jars.packages", "JohnSnowLabs:spark-nlp:2.0.1")\
    .config("spark.kryoserializer.buffer.max", "500m")\
    .getOrCreate()<jupyter_output><empty_output><jupyter_text>3. Load a spark dataset and put it in memory<jupyter_code>#Load the input data to be annotated
data = spark. \
        read. \
        parquet( resource_path+"sentiment.parquet"). \
        limit(1000).cache()
data.show()<jupyter_output><empty_output><jupyter_text>4. Create the document assembler, which will put target text column into Annotation form<jupyter_code>### Define the dataframe
document_assembler = DocumentAssembler() \
            .setInputCol("text")\
            .setOutputCol("document")

### Example: Checkout the output of document assembler
assembled = document_assembler.transform(data)
assembled.show(5)<jupyter_output><empty_output><jupyter_text>5. Create Sentence detector to parse sub sentences in every document<jupyter_code>### Sentence detector
sentence_detector = SentenceDetector() \
    .setInputCols(["document"]) \
    .setOutputCol("sentence")
### Example: Checkout the output of sentence detector
sentence_data = sentence_detector.transform(assembled)
sentence_data.show(5)<jupyter_output><empty_output><jupyter_text>6. The tokenizer will match standard tokens<jupyter_code>### Tokenizer
tokenizer = Tokenizer() \
            .setInputCols(["sentence"]) \
            .setOutputCol("token")

### Example: Checkout the outout of tokenizer
tokenized = tokenizer.transform(sentence_data)
tokenized.show(5)<jupyter_output><empty_output><jupyter_text>7. Normalizer will clean out the tokens<jupyter_code>normalizer = Normalizer() \
            .setInputCols(["token"]) \
            .setOutputCol("normal")<jupyter_output><empty_output><jupyter_text>8. The spell checker will correct normalized tokens, this trains with a dictionary of english words<jupyter_code>### Spell Checker
spell_checker = NorvigSweetingApproach() \
            .setInputCols(["normal"]) \
            .setOutputCol("spell") \
            .setDictionary( resource_path+ "spell/words.txt")<jupyter_output><empty_output><jupyter_text>9. Create the ViveknSentimentApproach and set resources to train it<jupyter_code>sentiment_detector = ViveknSentimentApproach() \
    .setInputCols(["spell", "sentence"]) \
    .setOutputCol("sentiment") \
    .setPruneCorpus(0) \
    .setPositiveSource(resource_path+"vivekn/positive") \
    .setNegativeSource(resource_path+"vivekn/negative") \<jupyter_output><empty_output><jupyter_text>10. The finisher will utilize sentiment analysis output<jupyter_code>finisher = Finisher() \
    .setInputCols(["sentiment"]) \
    .setIncludeMetadata(False)<jupyter_output><empty_output><jupyter_text>11. Fit and predict over data<jupyter_code>pipeline = Pipeline(stages=[
    document_assembler,
    sentence_detector,
    tokenizer,
    normalizer,
    spell_checker,
    sentiment_detector,
    finisher
])

start = time.time()
sentiment_data = pipeline.fit(data).transform(data)

end = time.time()
print("Time elapsed pipeline process: " + str(end - start))<jupyter_output><empty_output><jupyter_text>13. Check the result<jupyter_code>sentiment_data.show(5,False)
type(sentiment_data)

# Negative Sentiments
for r in sentiment_data.where(array_contains(sentiment_data.finished_sentiment, "negative")).take(5):
    print(r['text'].strip(),"->",r['finished_sentiment'])
# Positive Sentiments
for r in sentiment_data.where(array_contains(sentiment_data.finished_sentiment, "positive")).take(5):
    print(r['text'].strip(),"->",r['finished_sentiment'])<jupyter_output><empty_output><jupyter_text>14. Can also be used directly on an array of dummy text<jupyter_code>dummy_data = spark.sparkContext.parallelize([["I am happy and like this spark NLP"], ["Have to say something bad now"]]).toDF().toDF("text")
dummy_data.show()
pipeline.fit(dummy_data).transform(dummy_data).show()<jupyter_output><empty_output><jupyter_text>14. The pipeline could be saved on disk for future reuse. Either after or before fitting the model<jupyter_code>new_pipeline = Pipeline(stages=[
    document_assembler,
    sentence_detector,
    tokenizer,
    normalizer,
    spell_checker,
    sentiment_detector,
    finisher
])

start = time.time()
new_pipeline.write().overwrite().save("./ps")
end = time.time()
print("Time elapsed in write before fiting pipelines: " + str(end - start))
start = time.time()
new_pipeline.fit(data).write().overwrite().save("./ms")
end = time.time()
print("Time elapsed in write after fiting pipelines: " + str(end - start))<jupyter_output><empty_output><jupyter_text>15. Pipelines can be easily loaded back in memory<jupyter_code>start = time.time()
p = Pipeline.read().load("./ps")
pm = PipelineModel.read().load("./ms")
end = time.time()
print("Time elapsed in read pipelines: " + str(end - start))
# Using the fitted pipeline read from disk
start = time.time()
data_transformed=pm.transform(data)
data_transformed.where(array_contains(data_transformed.finished_sentiment, "negative")).show()
print(data_transformed.count())
end = time.time()
print("Time elapsed in using loaded pipelines: " + str(end - start))<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11844457687723481, 'fraction_numerical': 0.007896305125148988, 'mean_word_length': 5.436241610738255, 'pattern_counts': {'":': 0, '<': 61, '<?xml version=': 0, '>': 63, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '15', 'dir': 'jupyter-structured-clean-dedup', 'id': '639266', 'license': 'Apache-2.0', 'n_tokens_mistral': 2149, 'n_tokens_neox': 2079, 'n_words': 497, 'path': 'python/example/vivekn-sentiment/sentiment.ipynb', 'repo_name': 'picachu93/spark-nlp', 'size': '6712'}