<jupyter_start><jupyter_text>Exercise 2: Advanced Analytics NLP<jupyter_code>!pip install spark-nlp==1.7.3
import pandas as pd
pd.set_option('max_colwidth', 800)<jupyter_output><empty_output><jupyter_text>Create a spark context that includes a 3rd party jar for NLP<jupyter_code>#jarPath = "spark-nlp-assembly-1.7.3.jar"

from pyspark.sql import SparkSession
spark = SparkSession.builder \
    .config("spark.jars.packages", "JohnSnowLabs:spark-nlp:1.8.2") \
    .getOrCreate()
spark<jupyter_output><empty_output><jupyter_text>Read multiple files in a dir as one Dataframe<jupyter_code>dataPath = "./data/reddit/*.json"
df = spark.read.json(dataPath)
print(df.count())
df.printSchema()<jupyter_output>100
root
 |-- data: struct (nullable = true)
 |    |-- approved_at_utc: string (nullable = true)
 |    |-- approved_by: string (nullable = true)
 |    |-- archived: boolean (nullable = true)
 |    |-- author: string (nullable = true)
 |    |-- author_flair_background_color: string (nullable = true)
 |    |-- author_flair_css_class: string (nullable = true)
 |    |-- author_flair_richtext: array (nullable = true)
 |    |    |-- element: string (containsNull = true)
 |    |-- author_flair_template_id: string (nullable = true)
 |    |-- author_flair_text: string (nullable = true)
 |    |-- author_flair_text_color: string (nullable = true)
 |    |-- author_flair_type: string (nullable = true)
 |    |-- author_fullname: string (nullable = true)
 |    |-- author_patreon_flair: boolean (nullable = true)
 |    |-- banned_at_utc: string (nullable = true)
 |    |-- banned_by: string (nullable = true)
 |    |-- can_gild: boolean (nullable = true)
 |    |-- can_mod_post: boolean (nullable = true)
[...]<jupyter_text>Deal with Struct type to query subfields<jupyter_code>title = "data.title"
author = "data.author"
dfAuthorTilte = df.select(title, author)
dfAuthorTilte.limit(5).toPandas()<jupyter_output><empty_output><jupyter_text>Try to implement the equivalent of flatMap in dataframes<jupyter_code>import pyspark.sql.functions as F

dfWordCount = df.select(F.explode(F.split(title,"\\s+")).alias("word")).groupBy("word").count().orderBy(F.desc("count"))
dfWordCount.limit(10).toPandas()<jupyter_output><empty_output><jupyter_text>Use an NLP libary to do Part-of-Speech Tagging<jupyter_code>from com.johnsnowlabs.nlp.pretrained.pipeline.en import BasicPipeline as bp
dfAnnotated = bp.annotate(dfAuthorTilte, "title")
dfAnnotated.printSchema()<jupyter_output>root
 |-- text: string (nullable = true)
 |-- author: string (nullable = true)
 |-- document: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- annotatorType: string (nullable = true)
 |    |    |-- begin: integer (nullable = false)
 |    |    |-- end: integer (nullable = false)
 |    |    |-- result: string (nullable = true)
 |    |    |-- metadata: map (nullable = true)
 |    |    |    |-- key: string
 |    |    |    |-- value: string (valueContainsNull = true)
 |-- token: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- annotatorType: string (nullable = true)
 |    |    |-- begin: integer (nullable = false)
 |    |    |-- end: integer (nullable = false)
 |    |    |-- result: string (nullable = true)
 |    |    |-- metadata: map (nullable = true)
 |    |    |    |-- key: string
 |    |    |    |-- value: string (valueContainsNull = true)
 |-- normal: array (nullable = true)
 |    |-- element: struct (contains[...]<jupyter_text>Deal with Map type to query subfields<jupyter_code>dfPos = dfAnnotated.select("text", "pos.metadata", "pos.result")
dfPos.limit(5).toPandas()
dfPos= dfAnnotated.select(F.explode("pos").alias("pos"))
dfPos.printSchema()
dfPos.toPandas()<jupyter_output>root
 |-- pos: struct (nullable = true)
 |    |-- annotatorType: string (nullable = true)
 |    |-- begin: integer (nullable = false)
 |    |-- end: integer (nullable = false)
 |    |-- result: string (nullable = true)
 |    |-- metadata: map (nullable = true)
 |    |    |-- key: string
 |    |    |-- value: string (valueContainsNull = true)<jupyter_text>Keep only proper nouns NNP or NNPS<jupyter_code>nnpFilter = "pos.result = 'NNP' or pos.result = 'NNPS' "
dfNNP = dfPos.where(nnpFilter)
dfNNP.limit(10).toPandas()<jupyter_output><empty_output><jupyter_text>Extract columns form a map in a col<jupyter_code>dfWordTag = dfNNP.selectExpr("pos.metadata['word'] as word", "pos.result as tag")
dfWordTag.limit(10).toPandas()
from pyspark.sql.functions import desc
dfWordTag.groupBy("word").count().orderBy(desc("count")).show()<jupyter_output>+--------+-----+
|    word|count|
+--------+-----+
|      US|   14|
|   Trump|    9|
|   Saudi|    8|
|   Putin|    7|
|  Russia|    6|
|  Europe|    5|
|  Arabia|    5|
|Catholic|    4|
|      UK|    4|
|Vladimir|    4|
| Germany|    3|
|   China|    3|
|    Pope|    3|
|   Egypt|    3|
|   South|    3|
|   House|    3|
|  Church|    3|
|  Africa|    2|
|National|    2|
|  Health|    2|
+--------+-----+
only showing top 20 rows
{'fraction_non_alphanumeric': 0.1662674650698603, 'fraction_numerical': 0.009580838323353293, 'mean_word_length': 3.395614035087719, 'pattern_counts': {'":': 0, '<': 33, '<?xml version=': 0, '>': 33, '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': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '74094', 'license': 'MIT', 'n_tokens_mistral': 1818, 'n_tokens_neox': 1828, 'n_words': 375, 'path': 'data_lakes_w_spark/05_introduction_into_data_lakes/solutions/Exercise 2 - Advanced Analytics NLP - Solution.ipynb', 'repo_name': 'MonNum5/udacity_data_engineer', 'size': '5010'}