# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3
#     language: python
#     name: python3
# ---

# + [markdown] id="sXatvRX899i0"
# ![JohnSnowLabs](https://nlp.johnsnowlabs.com/assets/images/logo.png)

# + [markdown] id="Lt-CiWRewNWD"
# [![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/Public/6.Playground_DataFrames.ipynb)

# + [markdown] id="-Alh8i-_fJ59"
# # Spark DataFrames Playground

# + id="sJ0MnF3bfWbe"
# ! pip install -q spark-nlp==3.4.2 pyspark==3.2.0

# + [markdown] id="4ZucC6lQ2pYQ"
#
# <b>  if you want to work with Spark 2.3 </b>
# ```
# import os
#
# # Install java
# # # ! apt-get update -qq
# # # ! apt-get install -y openjdk-8-jdk-headless -qq > /dev/null
#
# # # !wget -q https://archive.apache.org/dist/spark/spark-2.3.0/spark-2.3.0-bin-hadoop2.7.tgz
#
# # # !tar xf spark-2.3.0-bin-hadoop2.7.tgz
# # # !pip install -q findspark
#
# os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"
# os.environ["PATH"] = os.environ["JAVA_HOME"] + "/bin:" + os.environ["PATH"]
# os.environ["SPARK_HOME"] = "/content/spark-2.3.0-bin-hadoop2.7"
# # # ! java -version
#
# import findspark
# findspark.init()
# from pyspark.sql import SparkSession
#
# # # ! pip install --ignore-installed -q spark-nlp==2.7.5
#
# import sparknlp
#
# spark = sparknlp.start(spark23=True)
# ```

# + colab={"base_uri": "https://localhost:8080/", "height": 254} outputId="c1aaeaa5-868c-444c-b4cc-f2e78ddf2996" id="pB-mZFa4O8ct" executionInfo={"status": "ok", "timestamp": 1649101837409, "user_tz": -180, "elapsed": 52138, "user": {"displayName": "Monster C", "userId": "08787989274818793476"}}
import sparknlp
import pandas as pd

from pyspark.ml import Pipeline
from pyspark.ml import PipelineModel

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

spark = sparknlp.start(spark32=True) # for GPU training >> sparknlp.start(gpu = True) # for Spark 2.3 =>> sparknlp.start(spark23 = True)

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

spark

# + id="JLulYEvRfDhG"
document = DocumentAssembler()\
    .setInputCol('text')\
    .setOutputCol('document')

# + id="tLfu1NU_fDhJ"
tokenizer = Tokenizer()\
    .setInputCols('document')\
    .setOutputCol('token')

# + id="HxZ9s3YCfDhM" colab={"base_uri": "https://localhost:8080/"} outputId="ff627c2e-8d66-41f6-afe4-1c4e83be48b0"
pos = PerceptronModel.pretrained()\
    .setInputCols(['document', 'token'])\
    .setOutputCol('pos')

# + id="cI-uo9ZbfDhR"
pipeline = Pipeline().setStages([
    document, 
    tokenizer, 
    pos])

# + id="MJoifWwRfjrG"
# !wget https://raw.githubusercontent.com/JohnSnowLabs/spark-nlp-workshop/master/jupyter/annotation/english/spark-nlp-basics/sample-sentences-en.txt

# + id="zGksf1hCfDhU" executionInfo={"status": "ok", "timestamp": 1649101840615, "user_tz": -180, "elapsed": 2742, "user": {"displayName": "Monster C", "userId": "08787989274818793476"}}
data = spark.read.text('./sample-sentences-en.txt').toDF('text')

# + id="KgTNu0_KfDhX" colab={"base_uri": "https://localhost:8080/"} outputId="6f78c799-970a-4d11-ad76-670c711e278a" executionInfo={"status": "ok", "timestamp": 1649101845958, "user_tz": -180, "elapsed": 3152, "user": {"displayName": "Monster C", "userId": "08787989274818793476"}}
data.show(5, truncate=False)

# + id="PAnijYcYfDhb"
model = pipeline.fit(data)

# + id="5_fTTzUPfDhd"
result = model.transform(data)

# + id="ll7rPvWefDhg" colab={"base_uri": "https://localhost:8080/"} outputId="7efc01c0-b901-4167-a3dd-247342279d68"
result.show(5)

# + id="i06WW8wzfDhk"
stored = result\
  .select('text', 'pos.begin', 'pos.end', 'pos.result', 'pos.metadata')\
  .toDF('text', 'pos_begin', 'pos_end', 'pos_result', 'pos_meta')\
  .cache()

# + id="MBQWLPjzfDhp" colab={"base_uri": "https://localhost:8080/"} outputId="3e64a114-c775-452c-b2e9-c04ace34cf78"
stored.printSchema()

# + id="X93ASKmGfDhw" colab={"base_uri": "https://localhost:8080/"} outputId="c0b9b134-db3d-4989-a609-6c86fdeecd76"
stored.show(5)

# + [markdown] id="rI8rO1GjfDhz"
# ---------
# ## Spark SQL Functions

# + id="5c5OVnNafDh0"
from pyspark.sql.functions import *

# + id="f_nWknqlfDh3" colab={"base_uri": "https://localhost:8080/"} outputId="dbce3274-7e4f-4f80-f11e-dd2a18bd91bd"
stored.filter(array_contains('pos_result', 'VBD')).show(5)

# + id="WwBH_f-1fDh7" colab={"base_uri": "https://localhost:8080/"} outputId="92ced3d6-2fd4-4619-ef8f-8612c45018e2"
stored.withColumn('token_count', size('pos_result')).select('pos_result', 'token_count').show(5)

# + id="CZn-kEFifDh_" colab={"base_uri": "https://localhost:8080/"} outputId="38af3b92-408f-4799-bfec-9fed57721e4f"
stored.select('text', array_max('pos_end')).show(5)

# + id="GfzcYDcFfDiC" colab={"base_uri": "https://localhost:8080/"} outputId="7474844d-88e0-4621-f4f4-24dbc048bfa0"
stored.withColumn('unique_pos', array_distinct('pos_result')).select('pos_result', 'unique_pos').show(5)

# + id="W9k5hwUSfDiE" colab={"base_uri": "https://localhost:8080/"} outputId="92af2638-5579-45cb-920d-9564d168a932"
stored.groupBy(array_distinct('pos_result')).count().show(10)

# + [markdown] id="5O_ERu47fDiI"
# ## SQL Functions with `col`

# + id="BP2dz_BqfDiJ"
from pyspark.sql.functions import col

# + id="H9a1KaVEfDiM" colab={"base_uri": "https://localhost:8080/"} outputId="4bad3bc0-66be-4b16-9ff0-be3e2bbed4c3"
stored.select(col('pos_meta').getItem(0).getItem('word')).show(5)

# + [markdown] id="B0TtaUgUfDiP"
# -------------
# ## Spark NLP Annotation UDFs

# + id="gd9sv5LcxEDf"
import pandas as pd
from sparknlp.functions import *
from pyspark.sql.types import ArrayType, StringType

# + id="Psqxd7eWfDiQ" colab={"base_uri": "https://localhost:8080/"} outputId="1a6944ad-1c65-4fbd-b4aa-ac11ee24ec39"
result.select('pos').show(1, truncate=False)


# + id="miYDmJiSfDiU"
@udf( StringType())
def nn_annotation(res,meta):
    nn = []
    for i,j in zip(res,meta):
      if i == "NN":
        nn.append(j["word"])
    return nn    


# + id="GeRbqnsiMAGP" outputId="547cef1f-003e-4db7-f79e-4cdb7d362f5d" colab={"base_uri": "https://localhost:8080/"}
result.withColumn("nn_tokens", nn_annotation(col("pos.result"), col("pos.metadata")))\
      .select("nn_tokens")\
      .show(truncate=False)

StarcoderdataJupyterScriptsDedupFiltered