<jupyter_start><jupyter_text>Twitter Disaster Detection Natural Language Processing at Scale using PySpark--- <NAME>Data provided by KaggleSpecial Thanks to IBM for Creating and Maintenaning the Course IBM Coursera Advanced Data Science Capstone  Introduction to the **Kaggle** competition  Detecting Disasters from Human Language is Hard            On plus side LOOK AT THE SKY LAST NIGHT IT WAS ABLAZE pic.twitter.com/qqsmshaJ3N&mdash; <NAME> (@AnyOtherAnnaK) August 6, 2015                 Wild Fire Season is back. The moorland on Winter Hill at the head of the Douglas Valley is ablaze this afternoon pic.twitter.com/4vyaytrx1b&mdash; <NAME> (@Wingate_Wood) March 27, 2020            Which Tweet is about a real disaster?  **Using a Set of 7,613 Pre-Labelled Tweets is it Possible to Predict Whether an Unkonwn Tweet is About a Disaster or not?**  Training Data is Provided by **Kaggle** in Tabular Form                  id      keyword      location      text      target                  0      1      None      None      Our Deeds are the Reason of this earthquake M...      1              1      4      None      None      Forest fire near La Ronge Sask. Canada      1              2      5      None      None      All residents asked to 'shelter in place' are ...      1              3      6      None      None      13,000 people receive wildfires evacuation or...      1              4      7      None      None      Just got sent this photo from Ruby Alaska as ...      1              5      8      None      None      RockyFire Update =&gt; California Hwy. 20 closed...      1        Using a Naive Bayes Model we are Able to Classify Almost 80% of all Tweets Correctly The Following Tasks Were Needed to Achieve the Result1. Dropping features with unexplainable correlation to target (e.g. keywords)2. Dropping stop words and standardizing tokens (e.g. runs → run)3. Transforming sentences to numerical features (word count, TF-IDF, and word2vec)4. Training multiple models with different hyper-parameters5. Selecting the best model6. Submitting the predictions to **Kaggle** and receive a score  Tweet Classification is Successfully Applied in Many Use Cases* Mapping territorial changes in the Syirian civil war ()* Predicting election results ()* Aiding disaster response ()  Details of my Solution  Architectural Considerations  Using **IBM Watson Studio** all Ressources to Kick-Start your Data Project are Ready to Use![Solution Arichitecture](./imgs/Architecture.png)  Downloading and Extracting the Data  Setting Up Spark Context<jupyter_code>import pyspark
from pyspark import SparkContext, SparkConf
from pyspark.sql import SparkSession

pyspark.__version__
sc = SparkContext.getOrCreate(SparkConf().setMaster("local[*]"))

spark = SparkSession \
    .builder \
    .getOrCreate()<jupyter_output><empty_output><jupyter_text>Retrieving Secrets Stored in Environment Variables<jupyter_code>import os
import getpass

def get_or_set_environment_variable(variable):
    try:
        var = os.environ[variable]
    except KeyError:
        var = getpass.getpass('Please enter value for {:}: '.format(variable))
    
    os.environ[variable] = var
    return var

ibm_api_key_id = get_or_set_environment_variable('IBM_API_KEY_ID')
ibm_cloud_store_bucket = get_or_set_environment_variable('IBM_OBJECT_STORE_BUCKET')<jupyter_output><empty_output><jupyter_text>Retrieve Credentials from Object Store …<jupyter_code>import json
import os

kaggle_path = os.path.join(os.curdir, 'work', 'kaggle.json')
with open(kaggle_path, 'r') as f_cr:
    creds = json.load(f_cr)

os.environ['KAGGLE_USERNAME'] = creds['username']
os.environ['KAGGLE_KEY'] = creds['key']
import types
from botocore.client import Config
import ibm_boto3

def __iter__(self): return 0

client = ibm_boto3.client(service_name='s3',
    ibm_api_key_id=ibm_api_key_id,
    ibm_auth_endpoint="https://iam.cloud.ibm.com/oidc/token",
    config=Config(signature_version='oauth'),
    endpoint_url='https://s3-api.us-geo.objectstorage.service.networklayer.com')

body = client.get_object(Bucket=ibm_cloud_store_bucket, Key='kaggle.json')['Body']
# add missing __iter__ method 

if not hasattr(body, "__iter__"): body.__iter__ = types.MethodType( __iter__, body )<jupyter_output><empty_output><jupyter_text>… and Store Them in Environment Variables<jupyter_code>creds = json.load(body)
os.environ['KAGGLE_USERNAME'] = creds['username']
os.environ['KAGGLE_KEY'] = creds['key']<jupyter_output><empty_output><jupyter_text>Finally, We Can Access **Kaggle** and List Current Competitions<jupyter_code>!kaggle competitions list<jupyter_output>ref                                            deadline             category            reward  teamCount  userHasEntered  
---------------------------------------------  -------------------  ---------------  ---------  ---------  --------------  
contradictory-my-dear-watson                   2030-07-01 23:59:00  Getting Started     Prizes         92           False  
gan-getting-started                            2030-07-01 23:59:00  Getting Started     Prizes        179           False  
tpu-getting-started                            2030-06-03 23:59:00  Getting Started  Knowledge        379           False  
digit-recognizer                               2030-01-01 00:00:00  Getting Started  Knowledge       2401           False  
titanic                                        2030-01-01 00:00:00  Getting Started  Knowledge      19019           False  
house-prices-advanced-regression-techniques    2030-01-01 00:00:00  Getting Started  Knowledge       5088           False  
[...]<jupyter_text>Downloading Data from **Kaggle**<jupyter_code>!kaggle competitions download -c nlp-getting-started -p "./work"
import glob

work_path = os.path.join(os.path.curdir, 'work')

zip_path = glob.glob(os.path.join(work_path, '*.zip'))[0]
zip_path<jupyter_output><empty_output><jupyter_text>Extracting Data Files from Archive<jupyter_code>import zipfile

with zipfile.ZipFile(zip_path, 'r') as zip_ref:
    zip_ref.extractall(work_path)
glob.glob(os.path.join(work_path, '*.csv'))<jupyter_output><empty_output><jupyter_text>Reading Training Data into Spark `DataFrame`<jupyter_code>df = spark.read.options(header=True,
                        inferschema=True,
                        multiLine=True,
                        encoding='UTF-8').csv(os.path.join(work_path, 'train.csv'))
df.sample(fraction=0.001).toPandas()<jupyter_output><empty_output><jupyter_text>Exploring the Data  Checking the Schema of the `DataFrame`<jupyter_code>df.printSchema()<jupyter_output>root
 |-- id: integer (nullable = true)
 |-- keyword: string (nullable = true)
 |-- location: string (nullable = true)
 |-- text: string (nullable = true)
 |-- target: integer (nullable = true)<jupyter_text>Getting a Feeling for the Data Using Summary Statisticts<jupyter_code>summary_statistics = df.describe()
summary_statistics.limit(20).toPandas()<jupyter_output><empty_output><jupyter_text>Calculating the Percentage of Filled Cells per Column<jupyter_code>row_count = df.count()
row_count
import pyspark.sql.functions as sfun
counts = df.select(sfun.count('id'),
                   sfun.count('keyword'),
                   sfun.count('location'),
                   sfun.count('text'),
                   sfun.count('target')).first()
per_filled = [cnt / float(row_count) for cnt in counts]
per_filled<jupyter_output><empty_output><jupyter_text>A Picture Says More than Thousand Words<jupyter_code>import plotly.io as pio
pio.renderers.default = 'notebook_connected'
import plotly.express as px
import pandas as pd

fig = px.bar(pd.DataFrame({'column': df.columns, 'filled': per_filled}),
             x='column', y='filled',
             title='Percentage of filled rows per column',
             text=counts,
             labels={'text': 'count'}
            )
fig.update_yaxes(tickformat=".02%")
fig.show()<jupyter_output><empty_output><jupyter_text>Let's Have a Look at the Top 10 Keywords per Target …<jupyter_code>top10_keywords = df.select('keyword').groupby('keyword').count().sort(sfun.desc('count')).na.drop().limit(10)
top10_with_target = df.select('target', 'keyword').groupby(
    'target', 'keyword').count().join(top10_keywords.select('keyword', sfun.col('count').alias('total')),
                                      on='keyword', how='right')
top10_with_target_px = top10_with_target.select('keyword', 'count', sfun.col('target').cast('String')).withColumn(
    'percentage', sfun.col('count')/float(row_count)).sort(sfun.asc('target'), sfun.desc('total'))

top10_keywords.toPandas()<jupyter_output><empty_output><jupyter_text>… and Plot them<jupyter_code>fig = px.bar(top10_with_target_px.toPandas(),
             x='keyword', y='percentage',
             color='target',
             title='Top 10 Keywords as percentage of total rows',
             text='count'
            )
fig.update_yaxes(tickformat=".02%")
fig.show()<jupyter_output><empty_output><jupyter_text>Now for the Locations<jupyter_code>top10_locations = df.select('location').groupby('location').count().orderBy(sfun.desc('count')).na.drop().limit(10)
top10_with_target = df.select('target', 'location').groupby(
    'target', 'location').count().join(top10_locations.select('location', sfun.col('count').alias('total')), on='location', how='right')
top10_with_target_px = top10_with_target.select('location', 'count', sfun.col('target').cast('String')).withColumn(
    'percentage', sfun.col('count')/float(row_count)).sort(sfun.asc('target'), sfun.desc('total'))

top10_locations.toPandas()
fig = px.bar(top10_with_target_px.toPandas(),
             x='location', y='percentage',
             color='target',
             title='Top 10 Locations as percentage of total rows',
             text='count'
            )
fig.update_yaxes(tickformat=".02%")
fig.show()<jupyter_output><empty_output><jupyter_text>What is the Percentage of Disastrous Tweets?<jupyter_code>tweets = df.select(sfun.col('target').cast('String')).groupby(
    'target').count().withColumn('column', sfun.lit('target')).sort('target').withColumn(
        'percentage', sfun.col('count')/row_count)

fig = px.bar(tweets.toPandas(),
             x='column', y='percentage',
             color='target',
             title='Target as percentage of total rows',
             text='count'
            )
fig.update_yaxes(tickformat=".02%")
fig.show()<jupyter_output><empty_output><jupyter_text>Let's Have a Look at Some Tweets That are not Describing a Disaster<jupyter_code>df.where(sfun.col('target') == 0).select('text').rdd.map(lambda row: row['text']).take(10)<jupyter_output><empty_output><jupyter_text>And This is How Disastrous Tweets Look<jupyter_code>df.where(sfun.col('target') == 1).select('text').rdd.map(lambda row: row['text']).take(10)<jupyter_output><empty_output><jupyter_text>Since We Are Using *Spark*, a Word Count is Mandatory<jupyter_code>words = df.select(sfun.lower(sfun.col('text')).alias('text'), 'target').rdd.flatMap(
    lambda row: [(word, (row['target'], 1 - row['target'])) for word in row['text'].split()])
wordcount = words.reduceByKey(lambda agg, cat: (agg[0] + cat[0], agg[1] + cat[1]))
top20_words = wordcount.sortBy(keyfunc=lambda row: -sum(row[1])).take(20)
top20_words<jupyter_output><empty_output><jupyter_text>A Cleanup of Stop Words is Necessary<jupyter_code>c_words, counts = zip(*top20_words)
c_words *= 2

count1, count0 = zip(*counts)
c_counts = count0 + count1

c_target = ['0'] * len(counts) + ['1'] * len(counts)

fig = px.bar(pd.DataFrame({'word': c_words,
                     'count': c_counts,
                     'target': c_target
                    }),
             x='word',
             y='count',
             title='Top 20 Words per Target',
             color='target')

fig.show()<jupyter_output><empty_output><jupyter_text>Extract – Transform – Load  Defining the Schema for Loading the `DataFrame` …<jupyter_code>from pyspark.sql.types import StructType, StructField, IntegerType, StringType

schema = StructType([StructField('id', IntegerType(), nullable=False),
                     StructField('keyword', StringType(), nullable=True),
                     StructField('location', StringType(), nullable=True),
                     StructField('text', StringType(), nullable=False)]
                   )
schema<jupyter_output><empty_output><jupyter_text>… and Loading the Data Using the `FAILFAST` Mode to Ensure Integrity<jupyter_code>df_test = spark.read.csv(os.path.join(work_path, 'test.csv'),
                         header=True,
                         encoding='UTF-8',
                         enforceSchema=True,
                         multiLine=True,
                         schema=schema,
                         mode='FAILFAST')
df_test.printSchema()
df_test.limit(10).toPandas()<jupyter_output>root
 |-- id: integer (nullable = true)
 |-- keyword: string (nullable = true)
 |-- location: string (nullable = true)
 |-- text: string (nullable = true)<jupyter_text>Amend the Schema with a `target` Column Before …<jupyter_code>schema.add(StructField('target', IntegerType(), nullable=False))
schema<jupyter_output><empty_output><jupyter_text>… Loading the Training Set<jupyter_code>df_train = spark.read.csv(os.path.join(work_path, 'train.csv'),
                          header=True,
                          encoding='UTF-8',
                          multiLine=True,
                          enforceSchema=True,
                          schema=schema,
                          mode='FAILFAST')
df_train.printSchema()
df_train.limit(10).toPandas()<jupyter_output>root
 |-- id: integer (nullable = true)
 |-- keyword: string (nullable = true)
 |-- location: string (nullable = true)
 |-- text: string (nullable = true)
 |-- target: integer (nullable = true)<jupyter_text>Finally, we Split Off the `target` Column to Ensure, that Both Training and Test Set Have an Identical Schema<jupyter_code>df_label = df_train.select('id', 'target')
df_label.printSchema()
df_label.limit(5).toPandas()
df_train = df_train.drop('target')<jupyter_output><empty_output><jupyter_text>Serializing the `DataFrame`-s in *Parquet* Format<jupyter_code>!rm -r ./disaster_detection_*
temp_parquet_file = os.path.join(os.path.curdir,
                                 'disaster_detection_{}')
df_train.write.parquet(temp_parquet_file.format('train'))
df_label.write.parquet(temp_parquet_file.format('label'))
df_test.write.parquet(temp_parquet_file.format('test'))

glob.glob(temp_parquet_file.format('*'))<jupyter_output><empty_output><jupyter_text>Uploading the Files to Object Cloud …<jupyter_code>def upload_parquet(client, path):
    parts = glob.glob(os.path.join(path, '*.parquet'))
    parquets = ['{:s}-{:04d}.parquet'.format(os.path.split(path)[-1], i)
                for i in range(len(parts))]
    for part, parquet in zip(parts, parquets):
        with open(part, 'rb') as parquetF:
            client.put_object(Bucket=ibm_cloud_store_bucket,
                          Body=parquetF,
                          Key=parquet
                         )
    return parquets

client = ibm_boto3.client(service_name='s3',
    ibm_api_key_id=ibm_api_key_id,
    ibm_auth_endpoint="https://iam.cloud.ibm.com/oidc/token",
    config=Config(signature_version='oauth'),
    endpoint_url='https://s3-api.us-geo.objectstorage.service.networklayer.com')


parquets = {}
for dataset in ('train', 'label', 'test'):
    parquets[dataset] = upload_parquet(client, temp_parquet_file.format(dataset))

print(parquets)<jupyter_output><empty_output><jupyter_text>… and Create a Data Catalogue for Easier Loading of Data<jupyter_code>import json

client.put_object(Bucket=ibm_cloud_store_bucket,
                  Body=json.dumps(parquets),
                  Key='etl_parquet_files.json')<jupyter_output><empty_output><jupyter_text>Data Cleansing  Download Data Catalogue from Object Store …<jupyter_code>import json
import os

import types
from botocore.client import Config
import ibm_boto3

def __iter__(self): return 0

client = ibm_boto3.client(service_name='s3',
    ibm_api_key_id=ibm_api_key_id,
    ibm_auth_endpoint="https://iam.cloud.ibm.com/oidc/token",
    config=Config(signature_version='oauth'),
    endpoint_url='https://s3-api.us-geo.objectstorage.service.networklayer.com')

body = client.get_object(Bucket=ibm_cloud_store_bucket,
                         Key='etl_parquet_files.json')['Body']
# add missing __iter__ method

if not hasattr(body, "__iter__"): body.__iter__ = types.MethodType( __iter__, body )

files = json.load(body)
files<jupyter_output><empty_output><jupyter_text>… and Use it to Retrieve the Data from Object Store<jupyter_code>def load_dataframe(files, **kargs):
    dfs = []
    for fn in files:
        body = client.get_object(Bucket=ibm_cloud_store_bucket,
                                 Key=fn)['Body']
        if not hasattr(body, "__iter__"):
            body.__iter__ = types.MethodType( __iter__, body )
        
        tfn = 'temp_{:}'.format(fn)
        with open(tfn, 'wb') as temp:
            temp.write(body.read())
        dfs.append(spark.read.options(**kargs).parquet(tfn))
    df = dfs.pop()
    for other in dfs:
        df = df.union(other)
    return df

df_train = load_dataframe(files['train'])
df_test = load_dataframe(files['test'])
df_label = load_dataframe(files['label'])<jupyter_output><empty_output><jupyter_text>Check if the Schemata Match<jupyter_code>df_train.schema == df_test.schema<jupyter_output><empty_output><jupyter_text>Drop Unused Columns<jupyter_code>def drop_unused_cols(df):
    return df.drop('location', 'keyword')

df_train = drop_unused_cols(df_train)
df_test = drop_unused_cols(df_test)

df_train.limit(10).toPandas()<jupyter_output><empty_output><jupyter_text>Define Functions for Cleansing of Text …<jupyter_code>!pip install unidecode
from unidecode import unidecode
import pyspark.sql.functions as sfun

@sfun.udf
def normalize_alphabet(text):
    text = unidecode(text)
    text = text.encode('ascii', errors='ignore').decode('utf-8', errors='ignore')
    text = text.lower()
    return text

def no_squares(text):
    return sfun.regexp_replace(text, '\[.*?\]', '')

def no_links(text):
    return sfun.regexp_replace(text, 'https?://\S+|www\.\S+', '')

def no_angles(text):
    return sfun.regexp_replace(text, '<.*?>+', '')

def no_punctuation(text):
    return sfun.regexp_replace(text, '[%s]' % re.escape(string.punctuation), '')

def no_newline(text):
    return sfun.regexp_replace(text, '\n', '')

def no_words_with_nums(text):
    return sfun.regexp_replace(text, '\w*\d\w*\s?', '')<jupyter_output><empty_output><jupyter_text>… and Apply Them to the `text` column<jupyter_code>import re
import string


def clean_text(text):    
    text = normalize_alphabet(text)
    text = no_squares(text)
    text = no_links(text)
    text = no_angles(text)
    text = no_punctuation(text)
    text = no_newline(text)
    text = no_words_with_nums(text)
    return text

df_train = df_train.withColumn('clean_text', clean_text('text'))
df_test = df_test.withColumn('clean_text', clean_text('text'))

df_train.limit(10).toPandas()<jupyter_output><empty_output><jupyter_text>Before we Can Filter Out Stop Words, we Need to Split the Text into Tokens<jupyter_code>df_train = df_train.select('id', 'text', sfun.split('clean_text', '\s+').alias('clean_text'))
df_test = df_test.select('id', 'text', sfun.split('clean_text', '\s+').alias('clean_text'))

df_train.limit(10).toPandas()<jupyter_output><empty_output><jupyter_text>Next we Download the Stopword Corpus<jupyter_code>import re

import nltk
from nltk.corpus import stopwords

nltk.download('stopwords')
stop = stopwords.words('english') + ["I'm", 'via', 'u'] + [word + "'s" for word in stopwords.words('english')]

stop = list(map(lambda word: re.sub('[{:}]'.format(re.escape(string.punctuation)), '', word).lower(),
                stop))

stop[:10]<jupyter_output>[nltk_data] Downloading package stopwords to
[nltk_data]     /Users/timbenediktherbstrith/nltk_data...
[nltk_data]   Package stopwords is already up-to-date!<jupyter_text>We Remove all Stopwords from the Text<jupyter_code>@sfun.udf
def remove_stopwords(text):
    return list(filter(lambda word: not word in stop, text))

df_train = df_train.select('id', 'text', remove_stopwords('clean_text').alias('clean_text'))
df_test = df_test.select('id', 'text', remove_stopwords('clean_text').alias('clean_text'))

df_train.limit(10).toPandas()<jupyter_output><empty_output><jupyter_text>We Lemmatize the Words to Ensure that the Machine Treats ‘fires’ and ‘fire’ as the Same Word<jupyter_code>from nltk.stem import WordNetLemmatizer

nltk.download('wordnet')

lemma = WordNetLemmatizer()

@sfun.udf
def lemmatize_words(words):
    def exhaustive_lemmatization(word):
        for pos in ['n', 'v', 'a']:
            lem = lemma.lemmatize(word, pos=pos)
            if lem != word:
                return lem
        return word
    return list(map(exhaustive_lemmatization, words))

df_train = df_train.select('id', 'text', lemmatize_words('clean_text').alias('clean_text'))
df_test = df_test.select('id', 'text', lemmatize_words('clean_text').alias('clean_text'))

df_train.limit(10).toPandas()<jupyter_output>[nltk_data] Downloading package wordnet to
[nltk_data]     /Users/timbenediktherbstrith/nltk_data...
[nltk_data]   Package wordnet is already up-to-date!<jupyter_text>Finally, we Join the Words to Form a Sentence Again<jupyter_code>@sfun.udf
def join_words(words):
    return ' '.join(words)

df_train = df_train.select('id', 'text', join_words('clean_text').alias('clean_text'))
df_test = df_test.select('id', 'text', join_words('clean_text').alias('clean_text'))

df_train.limit(10).toPandas()<jupyter_output><empty_output><jupyter_text>Let's Check if the Word Counts Have Become More Significant<jupyter_code>df = df_train.join(df_label, on='id', how='inner')

words = df.rdd.flatMap(
    lambda row: [(word, (int(row['target']), 1 - int(row['target']))) for word in row['clean_text'].split()])
wordcount = words.reduceByKey(lambda agg, cat: (agg[0] + cat[0], agg[1] + cat[1]))
print(wordcount.count())
top20_words = wordcount.sortBy(keyfunc=lambda row: -sum(row[1])).take(20)
top20_words<jupyter_output>14101<jupyter_text>Indeed, the Word Counts Reflects More Important Words after the Cleanup<jupyter_code>import pandas as pd
import plotly.express as px

c_words, counts = zip(*top20_words)
c_words *= 2

count1, count0 = zip(*counts)
c_counts = count0 + count1

c_target = ['0'] * len(counts) + ['1'] * len(counts)

fig = px.bar(pd.DataFrame({'word': c_words,
                     'count': c_counts,
                     'target': c_target
                    }),
             x='word',
             y='count',
             text='count',
             title='Top 20 Words per Target',
             color='target')

fig.show()<jupyter_output><empty_output><jupyter_text>Feature Engineering  We Create 3 Features from the Data and Will Evaluate Performance of the resulting Models* `CountVectorizer`: Converts a text document to a sparse vector of token counts.* `TF-IDF`: Short for ‘term frequency–inverse document frequency’, is a numerical statistic that is intended to reflect how important a word is. In short we count the occurences of the word per document and normalize the count by taking the overall frequency of the term into account.* `Word2Vec`: A neural network model is trained to learn word associations from a large corpus of text.  Creating the Features Using the *PySpark* API<jupyter_code>df_train = spark.read.parquet('disaster_detection_clean_train-0000.parquet')
df_test = spark.read.parquet('disaster_detection_clean_test-0000.parquet')
from pyspark.ml.feature import RegexTokenizer, CountVectorizer, HashingTF, IDF, Word2Vec
from pyspark.ml import Pipeline

regex_tokenizer = RegexTokenizer(inputCol='clean_text', outputCol='words', pattern=r'\W')
counter = CountVectorizer(inputCol='words', outputCol='features_count', vocabSize=2500, minDF=5)
hashing_tf = HashingTF(inputCol='words', outputCol='raw_features', numFeatures=2500)
idf = IDF(inputCol='raw_features', outputCol='features_tfidf', minDocFreq=5)
word2vec = Word2Vec(inputCol='words', outputCol='features_w2v', maxSentenceLength=50)

pipeline = Pipeline(stages=[regex_tokenizer,
                            counter,
                            hashing_tf,
                            idf,
                            word2vec
                            ]).fit(df_train)

df_train = pipeline.transform(df_train).select('id', 'text', 'features_count', 'features_tfidf', 'features_w2v')
df_test = pipeline.transform(df_test).select('id', 'text', 'features_count', 'features_tfidf', 'features_w2v')<jupyter_output><empty_output><jupyter_text>Let's have a Look at the Resulting Dataset<jupyter_code>assert df_train.schema == df_test.schema

df_train.show()<jupyter_output>+---+--------------------+--------------------+--------------------+--------------------+
| id|                text|      features_count|      features_tfidf|        features_w2v|
+---+--------------------+--------------------+--------------------+--------------------+
|  1|Our Deeds are the...|(2266,[16,80,201,...|(2500,[26,166,336...|[-0.0027080011953...|
|  4|Forest fire near ...|(2266,[2,139,183,...|(2500,[191,974,20...|[-0.0017992729055...|
|  5|All residents ask...|(2266,[212,318,32...|(2500,[294,691,11...|[0.00301898798947...|
|  6|13,000 people rec...|(2266,[8,41,85,21...|(2500,[8,325,644,...|[-0.0085309529677...|
|  7|Just got sent thi...|(2266,[0,85,130,1...|(2500,[178,198,32...|[-0.0092202152849...|
|  8|#RockyFire Update...|(2266,[2,41,85,21...|(2500,[191,325,64...|[-0.0043081726810...|
| 10|#flood #disaster ...|(2266,[15,21,57,1...|(2500,[353,356,64...|[0.00218058816002...|
| 13|I'm on top of the...|(2266,[2,25,187,1...|(2500,[15,191,187...|[-0.0103795815724...|
| 14|There[...]<jupyter_text>Model definition  Logistic Regression<jupyter_code>from pyspark.ml.classification import LogisticRegression

lrs = [LogisticRegression(featuresCol=feat,
                          maxIter=20, regParam=0.3, elasticNetParam=0)
       for feat in ('features_count', 'features_tfidf')]
lrs<jupyter_output><empty_output><jupyter_text>Naive Bayes<jupyter_code>from pyspark.ml.classification import NaiveBayes

nbs = [NaiveBayes(featuresCol=feat, smoothing=1)
       for feat in ('features_count', 'features_tfidf')]
nbs<jupyter_output><empty_output><jupyter_text>Convolutional Neural Network<jupyter_code>import tensorflow as tf
tf.__version__
from tensorflow.keras.layers import Dense, Dropout, LeakyReLU
from tensorflow.keras.models import Sequential
from tensorflow.keras import optimizers

MAX_SEQUENCE_LENGTH = len(df_train.select('features_w2v').first()['features_w2v'])

model = Sequential()
model.add(Dense(82, input_dim=MAX_SEQUENCE_LENGTH))
model.add(LeakyReLU(alpha=0.01))
model.add(Dropout(0.20))
model.add(Dense(82))
model.add(LeakyReLU(alpha=0.01))
model.add(Dropout(0.20))
model.add(Dense(1, activation='sigmoid'))
model.compile(loss='binary_crossentropy', optimizer='adam', metrics=['accuracy'])<jupyter_output><empty_output><jupyter_text>Model Training  Transforming and Splitting Training Data<jupyter_code>df_train = df_train.join(df_label.select('id', sfun.col('target').alias('label')), on='id', how='inner')
df_training, df_validation = df_train.randomSplit([0.7, 0.3], seed=42)
print("Training Dataset Count: " + str(df_training.count()))
print("Test Dataset Count: " + str(df_validation.count()))<jupyter_output>Training Dataset Count: 5355
Test Dataset Count: 2258<jupyter_text>Training *Spark ML* Models<jupyter_code>spark_models = lrs + nbs

def training(model):
    print('Training', model)
    return model.fit(df_training)

trained = [training(model) for model in spark_models]
trained<jupyter_output>Training LogisticRegression_62e5c1c34a57
Training LogisticRegression_6c13f1c946ca
Training NaiveBayes_9281f4601e98
Training NaiveBayes_51905823d4ab<jupyter_text>Transforming `DataFrame`-s to `numpy.array`<jupyter_code>import numpy as np
X = np.array(df_training.select('features_w2v').collect())
X = X.reshape(-1, 100)

y = np.array(df_training.select('label').collect())
y = y.reshape(-1,)

X.shape<jupyter_output><empty_output><jupyter_text>Training the *Keras* Model<jupyter_code>history = model.fit(X, y,
                    batch_size=124, epochs=50, validation_split=0.2)<jupyter_output>Epoch 1/50
35/35 [==============================] - 1s 21ms/step - loss: 0.6775 - accuracy: 0.6076 - val_loss: 0.6507 - val_accuracy: 0.6695
Epoch 2/50
35/35 [==============================] - 1s 17ms/step - loss: 0.6389 - accuracy: 0.6587 - val_loss: 0.6057 - val_accuracy: 0.7040
Epoch 3/50
35/35 [==============================] - 0s 14ms/step - loss: 0.6205 - accuracy: 0.6767 - val_loss: 0.6005 - val_accuracy: 0.7199
Epoch 4/50
35/35 [==============================] - 0s 9ms/step - loss: 0.6133 - accuracy: 0.6755 - val_loss: 0.5940 - val_accuracy: 0.7152
Epoch 5/50
35/35 [==============================] - 0s 10ms/step - loss: 0.6014 - accuracy: 0.6795 - val_loss: 0.5872 - val_accuracy: 0.6928
Epoch 6/50
35/35 [==============================] - 0s 10ms/step - loss: 0.6000 - accuracy: 0.6821 - val_loss: 0.5879 - val_accuracy: 0.6928
Epoch 7/50
35/35 [==============================] - 1s 24ms/step - loss: 0.5972 - accuracy: 0.6895 - val_loss: 0.5856 - val_accuracy: 0.6956
Epoch 8/50
35/[...]<jupyter_text>Plotting Loss and Accuracy Curves of Training<jupyter_code>import plotly.express as px
import pandas as pd

loss_values = history.history['loss']
history.history['epoche'] = range(1, len(loss_values)+1)

fig = px.line(pd.DataFrame(history.history),
              x='epoche', y=['loss', 'accuracy', 'val_loss', 'val_accuracy'])
fig.show()<jupyter_output><empty_output><jupyter_text>Model Evaluation  Prediction with Spark Models<jupyter_code>spark_models = lrs + nbs

predictions = [s_model.transform(df_validation).select('id', 'text', 'label', 'probability', 'prediction')
               for s_model in trained]

predictions[0].limit(10).toPandas()<jupyter_output><empty_output><jupyter_text>Prediction with *Keras*<jupyter_code>import numpy as np

Xv = np.array(df_validation.select('features_w2v').collect())
Xv = Xv.reshape(-1, 100)
yv = np.array(df_validation.select('label').collect()).reshape(-1,)

ypr = model.predict(Xv)
ypv = np.round(ypr)

ypv<jupyter_output><empty_output><jupyter_text>Transforming the Predictions of *Keras* into a `DataFrame`<jupyter_code>prediction = spark.createDataFrame([(float(l), float(p), [float(1 - r), float(r)])
                                    for l, p, r in zip(yv, ypv, ypr)],
                                   schema=['label', 'prediction', 'probability'])
predictions.append(prediction)

prediction.limit(10).toPandas()<jupyter_output><empty_output><jupyter_text>Now We Can Evaluate all Models Using the same API<jupyter_code>from pyspark.ml.evaluation import MulticlassClassificationEvaluator
evaluatorMulti = MulticlassClassificationEvaluator(labelCol='label', predictionCol='prediction')
import pandas as pd

pd.DataFrame({'model': ['Logistic Regression'] * 2 + ['Naive Bayes'] * 2 + ['Convolutional Neural Network'],
              'features': ['Count', 'TF-IDF'] * 2 + ['Word2Vec'],
              'precision': [evaluatorMulti.evaluate(prediction, {evaluatorMulti.metricName: 'weightedPrecision'})
                            for prediction in predictions],
              'recall': [evaluatorMulti.evaluate(prediction, {evaluatorMulti.metricName: 'weightedRecall'})
                         for prediction in predictions],
              'accuracy': [evaluatorMulti.evaluate(prediction, {evaluatorMulti.metricName: 'accuracy'})
                           for prediction in predictions],
              'F1': [evaluatorMulti.evaluate(prediction, {evaluatorMulti.metricName: 'f1'})
                     for prediction in predictions]})
from pyspark.mllib.evaluation import BinaryClassificationMetrics

# Scala version implements .roc() and .pr()
# Python: https://spark.apache.org/docs/latest/api/python/_modules/pyspark/mllib/common.html
# Scala: https://spark.apache.org/docs/latest/api/java/org/apache/spark/mllib/evaluation/BinaryClassificationMetrics.html
class CurveMetrics(BinaryClassificationMetrics):
    def __init__(self, *args):
        super(CurveMetrics, self).__init__(*args)

    def _to_list(self, rdd):
        points = []
        # Note this collect could be inefficient for large datasets 
        # considering there may be one probability per datapoint (at most)
        # The Scala version takes a numBins parameter, 
        # but it doesn't seem possible to pass this from Python to Java
        for row in rdd.collect():
            # Results are returned as type scala.Tuple2, 
            # which doesn't appear to have a py4j mapping
            points += [(float(row._1()), float(row._2()))]
        return points

    def get_curve(self, method):
        rdd = getattr(self._java_model, method)().toJavaRDD()
        return self._to_list(rdd)

points = []
mods = []
for mod, prediction in zip(['Logistic Regression', 'Naive Bayes', 'Convolutional Neural Network'], predictions[::2]):
    preds = prediction.select('label','probability').rdd.map(lambda row: (float(row['probability'][1]), float(row['label'])))
    curve = CurveMetrics(preds).get_curve('roc')
    points += curve
    mods += [mod] * len(curve)

points[:20]
import pandas as pd

fpr, tpr = zip(*points)
sample = len(points) // 3
df_roc = pd.DataFrame({'True Positive Rate': tpr, 'False Positive Rate': fpr,
                       'Model': mods})
df_roc<jupyter_output><empty_output><jupyter_text>As a Final Measure We Consider the ROC-Curve<jupyter_code>import plotly.express as px

fig = px.line(df_roc, x='False Positive Rate', y='True Positive Rate', color='Model',
              title='Receiver Operating Characteristic')
fig.show()<jupyter_output><empty_output><jupyter_text>Model Deployment  Predicting the Test Data<jupyter_code>model = trained[2]
prediction = model.transform(df_test).select('id', 'text', 'prediction')

prediction.limit(10).toPandas()<jupyter_output><empty_output><jupyter_text>Exporting the Data in Specified Format<jupyter_code>from pyspark.sql.types import IntegerType
import pyspark.sql.functions as sfun

df_exp = prediction.select('id', sfun.col('prediction').cast(IntegerType()).alias('target'))
df_exp.limit(10).toPandas()
df_exp.toPandas().to_csv('submission.csv', index=False)<jupyter_output><empty_output><jupyter_text>Finally, We Submit the *CSV* to **Kaggle**<jupyter_code>!kaggle competitions submit -f submission.csv -m "Submitting from IBM cloud to Kaggle" nlp-getting-started<jupyter_output><empty_output><jupyter_text>Let's See How We Did<jupyter_code>!kaggle competitions submissions nlp-getting-started<jupyter_output>fileName        date                 description                          status    publicScore  privateScore  
--------------  -------------------  -----------------------------------  --------  -----------  ------------  
submission.csv  2021-02-06 21:57:44  Submitting from IBM cloud to Kaggle  complete  0.78700      None          
submission.csv  2021-02-06 16:49:36  Second valid submission              complete  0.78700      None          
submission.csv  2021-01-30 22:31:42  First valid submission               complete  0.78700      None          
submission.csv  2021-01-30 22:30:36  Submitting from IBM cloud to Kaggle  error     None         None
{'fraction_non_alphanumeric': 0.1456972291056544, 'fraction_numerical': 0.02982090560937148, 'mean_word_length': 3.757904608788853, 'pattern_counts': {'":': 0, '<': 233, '<?xml version=': 0, '>': 233, 'https://': 8, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 24, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '60', 'dir': 'jupyter-structured-clean-dedup', 'id': '456556', 'license': 'MIT', 'n_tokens_mistral': 12450, 'n_tokens_neox': 11615, 'n_words': 2406, 'path': 'presentation/disaster_detection.presentation.py37.ipynb', 'repo_name': 'tim6her/kaggle_disaster_detection', 'size': '35529'}