<jupyter_start><jupyter_text>Vivekn Sentiment analysis computes Vivek Nayamanan algorithm from a set of positive and negative corpus We call necessary imports<jupyter_code>#Imports
import time
import sys
import os
sys.path.append('../../')

from pyspark.ml import Pipeline, PipelineModel
from sparknlp.annotator import *
from sparknlp.base import DocumentAssembler, Finisher<jupyter_output><empty_output><jupyter_text>We load SparkSession if not already there<jupyter_code>spark = SparkSession.builder \
    .appName("ner")\
    .master("local[1]")\
    .config("spark.driver.memory","8G")\
    .config("spark.driver.maxResultSize", "2G")\
    .config("spark.jar", "lib/sparknlp.jar")\
    .config("spark.kryoserializer.buffer.max", "500m")\
    .getOrCreate()<jupyter_output><empty_output><jupyter_text>We load a spark dataset and put it in memory<jupyter_code>#Load the input data to be annotated
data = spark. \
        read. \
        parquet("file:///" + os.getcwd() + "/../../../src/test/resources/sentiment.parquet"). \
        limit(1000)
data.cache()
data.count()
data.show()<jupyter_output><empty_output><jupyter_text>We creat the document assemblerr, which will put target text column into Annotation form<jupyter_code>### Define the dataframe
document_assembler = DocumentAssembler() \
            .setInputCol("text")
        
### Transform input to appropriate schema
#assembled = document_assembler.transform(data)<jupyter_output><empty_output><jupyter_text>The sentence detector will parse sub sentences in every line<jupyter_code>### Sentence detector
sentence_detector = SentenceDetector() \
    .setInputCols(["document"]) \
    .setOutputCol("sentence")
#sentence_data = sentence_detector.transform(checked)<jupyter_output><empty_output><jupyter_text>The tokenizer will match standard tokens<jupyter_code>### Tokenizer
tokenizer = Tokenizer() \
            .setInputCols(["sentence"]) \
            .setOutputCol("token")
#tokenized = tokenizer.transform(assembled)<jupyter_output><empty_output><jupyter_text>Normalizer will clean out the tokens<jupyter_code>normalizer = Normalizer() \
            .setInputCols(["token"]) \
            .setOutputCol("normal")<jupyter_output><empty_output><jupyter_text>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("file:///" + os.getcwd() + "/../../../src/test/resources/spell/words.txt")

#checked = spell_checker.fit(tokenized).transform(tokenized)<jupyter_output><empty_output><jupyter_text>We creat the ViveknSentimentApproach and set resources to train it<jupyter_code>sentiment_detector = ViveknSentimentApproach() \
    .setInputCols(["spell", "sentence"]) \
    .setOutputCol("sentiment") \
    .setPruneCorpus(0) \
    .setPositiveSource("file:///" + os.getcwd() + "/../../../src/test/resources/vivekn/positive") \
    .setNegativeSource("file:///" + os.getcwd() + "/../../../src/test/resources/vivekn/negative") \<jupyter_output><empty_output><jupyter_text>The finisher will utilize sentiment analysis output<jupyter_code>finisher = Finisher() \
    .setInputCols(["sentiment"]) \
    .setIncludeKeys(True)<jupyter_output><empty_output><jupyter_text>Dummy training text<jupyter_code>training = spark.sparkContext.parallelize([["hello world"], ["this is some more text"], ["and here another sentence"]]).toDF().toDF("text")
training.show()<jupyter_output><empty_output><jupyter_text>We 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(training).transform(data)
sentiment_data.show()
end = time.time()
print("Time elapsed pipeline process: " + str(end - start))<jupyter_output><empty_output><jupyter_text>We can take a sample back into the driver<jupyter_code>for r in sentiment_data.take(5):
    print(r)<jupyter_output><empty_output><jupyter_text>We save it to disk and re read it. Either after or before fitting the model<jupyter_code>start = time.time()
pipeline.write().overwrite().save("./ps")
pipeline.fit(data).write().overwrite().save("./ms")
end = time.time()
print("Time elapsed in write pipelines: " + str(end - start))
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))
start = time.time()
pm.transform(data).where("finished_sentiment not like '%negative%'").show()
print(pm.transform(data).count())
end = time.time()
print("Time elapsed in using loaded pipelines: " + str(end - start))<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1445360824742268, 'fraction_numerical': 0.002474226804123711, 'mean_word_length': 4.974137931034483, 'pattern_counts': {'":': 0, '<': 57, '<?xml version=': 0, '>': 57, 'https://': 0, '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': '14', 'dir': 'jupyter-structured-clean-dedup', 'id': '71307', 'license': 'Apache-2.0', 'n_tokens_mistral': 1597, 'n_tokens_neox': 1559, 'n_words': 310, 'path': 'python/example/vivekn-sentiment/sentiment.ipynb', 'repo_name': 'lgaud/spark-nlp', 'size': '4850'}