<jupyter_start><jupyter_text>Machine Learning for Telecom with Naive Bayes  Introduction Machine Learning for CallDisconnectReason is a notebook which demonstrates exploration of dataset and CallDisconnectReason classification with Spark ml Naive Bayes Algorithm.<jupyter_code>from pyspark.sql.types import *
from pyspark.sql import SparkSession
from sagemaker import get_execution_role
import sagemaker_pyspark


role = get_execution_role()

# Configure Spark to use the SageMaker Spark dependency jars
jars = sagemaker_pyspark.classpath_jars()

classpath = ":".join(sagemaker_pyspark.classpath_jars())

spark = SparkSession.builder.config("spark.driver.extraClassPath", classpath)\
    .master("local[*]").getOrCreate()<jupyter_output><empty_output><jupyter_text>Using S3 Select, enables applications to retrieve only a subset of data from an object by using simple SQL expressions. By using S3 Select to retrieve only the data, you can achieve drastic performance increases – in many cases you can get as much as a 400% improvement.- _We first read a parquet compressed format of CDR dataset using s3select which has already been processed by Glue._<jupyter_code>cdr_start_loc = "<%CDRStartFile%>"
cdr_stop_loc = "<%CDRStopFile%>"
cdr_start_sample_loc = "<%CDRStartSampleFile%>"
cdr_stop_sample_loc = "<%CDRStopSampleFile%>"

df = spark.read.format("s3select").parquet(cdr_stop_sample_loc)
df.createOrReplaceTempView("cdr")
durationDF = spark.sql("SELECT _c13 as CallServiceDuration FROM cdr where _c0 = 'STOP'")
durationDF.count()<jupyter_output><empty_output><jupyter_text>Exploration of Data - _We see how we can explore and visualize the dataset used for processing. Here we create a bar chart representation of CallServiceDuration from CDR dataset._<jupyter_code>import matplotlib.pyplot as plt
durationpd = durationDF.toPandas().astype(int)   
durationpd.plot(kind='bar',stacked=True,width=1)<jupyter_output><empty_output><jupyter_text>- _We can represent the data and visualize with a box plot. The box extends from the lower to upper quartile values of the data, with a line at the median._<jupyter_code>color = dict(boxes='DarkGreen', whiskers='DarkOrange',
           medians='DarkBlue', caps='Gray')

durationpd.plot.box(color=color, sym='r+')
from pyspark.sql.functions import col
durationDF = durationDF.withColumn("CallServiceDuration", col("CallServiceDuration").cast(DoubleType()))<jupyter_output><empty_output><jupyter_text>- _We can represent the data and visualize the data with histograms partitioned in different bins._<jupyter_code>import matplotlib.pyplot as plt

bins, counts = durationDF.select('CallServiceDuration').rdd.flatMap(lambda x: x).histogram(durationDF.count())
plt.hist(bins[:-1], bins=bins, weights=counts,color=['green'])
sqlDF = spark.sql("SELECT _c2 as Accounting_ID, _c19 as Calling_Number,_c20 as Called_Number, _c14 as CallDisconnectReason FROM cdr where _c0 = 'STOP'")
sqlDF.show()<jupyter_output>+------------------+--------------+-------------+--------------------+
|     Accounting_ID|Calling_Number|Called_Number|CallDisconnectReason|
+------------------+--------------+-------------+--------------------+
|0x00016E0F5BDACAF7|    9645000046|   3512000046|                  16|
|0x00016E0F36A4A836|    9645000048|   3512000048|                  16|
|0x00016E0F4C261126|    9645000050|   3512000050|                  16|
|0x00016E0F4A446638|    9645000052|   3512000052|                  16|
|0x00016E0F4040CE81|    9645000054|   3512000054|                  16|
|0x00016E0F4D522D63|    9645000055|   3512000055|                  16|
|0x00016E0F5854A088|    9645000057|   3512000057|                  16|
|0x00016E0F7DFDA482|    9645000060|   3512000060|                  16|
|0x00016E0F65D65F76|    9645000062|   3512000062|                  16|
|0x00016E0F2378A4AE|    9645000064|   3512000064|                  16|
|0x00016E0F5003BC72|    9645000066|   3512000066|                  16|
| 0x00[...]<jupyter_text>Featurization<jupyter_code>from pyspark.ml.feature import StringIndexer

accountIndexer = StringIndexer(inputCol="Accounting_ID", outputCol="AccountingIDIndex")
accountIndexer.setHandleInvalid("skip")
tempdf1 = accountIndexer.fit(sqlDF).transform(sqlDF)

callingNumberIndexer = StringIndexer(inputCol="Calling_Number", outputCol="Calling_NumberIndex")
callingNumberIndexer.setHandleInvalid("skip")
tempdf2 = callingNumberIndexer.fit(tempdf1).transform(tempdf1)

calledNumberIndexer = StringIndexer(inputCol="Called_Number", outputCol="Called_NumberIndex")
calledNumberIndexer.setHandleInvalid("skip")
tempdf3 = calledNumberIndexer.fit(tempdf2).transform(tempdf2)
from pyspark.ml.feature import StringIndexer
# Convert target into numerical categories
labelIndexer = StringIndexer(inputCol="CallDisconnectReason", outputCol="label")
labelIndexer.setHandleInvalid("skip")
from pyspark.sql.functions import rand

trainingFraction = 0.75; 
testingFraction = (1-trainingFraction);
seed = 1234;

trainData, testData = tempdf3.randomSplit([trainingFraction, testingFraction], seed=seed);

# CACHE TRAIN AND TEST DATA
trainData.cache()
testData.cache()
trainData.count(),testData.count()<jupyter_output><empty_output><jupyter_text>Analyzing the label distribution- We analyze the distribution of our target labels using a histogram where 16 represents Normal_Call_Clearing.<jupyter_code>import matplotlib.pyplot as plt

negcount = trainData.filter("CallDisconnectReason != 16").count()
poscount = trainData.filter("CallDisconnectReason == 16").count()

negfrac = 100*float(negcount)/float(negcount+poscount)
posfrac = 100*float(poscount)/float(poscount+negcount)
ind = [0.0,1.0]
frac = [negfrac,posfrac]
width = 0.35

plt.title('Label Distribution')
plt.bar(ind, frac, width, color='r')
plt.xlabel("CallDisconnectReason")
plt.ylabel('Percentage share')
plt.xticks(ind,['0.0','1.0'])
plt.show()
import matplotlib.pyplot as plt

negcount = testData.filter("CallDisconnectReason != 16").count()
poscount = testData.filter("CallDisconnectReason == 16").count()

negfrac = 100*float(negcount)/float(negcount+poscount)
posfrac = 100*float(poscount)/float(poscount+negcount)
ind = [0.0,1.0]
frac = [negfrac,posfrac]
width = 0.35

plt.title('Label Distribution')
plt.bar(ind, frac, width, color='r')
plt.xlabel("CallDisconnectReason")
plt.ylabel('Percentage share')
plt.xticks(ind,['0.0','1.0'])
plt.show()
from pyspark.ml.feature import VectorAssembler

from pyspark.ml.feature import VectorAssembler

vecAssembler = VectorAssembler(inputCols=["AccountingIDIndex","Calling_NumberIndex", "Called_NumberIndex"], outputCol="features")<jupyter_output><empty_output><jupyter_text>__Spark ML Naive Bayes__:     Naive Bayes is a simple multiclass classification algorithm with the assumption of independence between every pair of features. Naive Bayes can be trained very efficiently. Within a single pass to the training data, it computes the conditional probability distribution of each feature given label, and then it applies Bayes’ theorem to compute the conditional probability distribution of label given an observation and use it for prediction.- _We use Spark ML Naive Bayes Algorithm and spark Pipeline to train the data set._<jupyter_code>from pyspark.ml.classification import NaiveBayes
from pyspark.ml.clustering import KMeans
from pyspark.ml import Pipeline

# Train a NaiveBayes model
nb = NaiveBayes(smoothing=1.0, modelType="multinomial")

# Chain labelIndexer, vecAssembler and NBmodel in a 
pipeline = Pipeline(stages=[labelIndexer,vecAssembler, nb])

# Run stages in pipeline and train model
model = pipeline.fit(trainData)
 # Run inference on the test data and show some results
predictions = model.transform(testData)
predictions.printSchema()
predictions.show()
predictiondf = predictions.select("label", "prediction", "probability")
pddf_pred = predictions.toPandas()
pddf_pred<jupyter_output><empty_output><jupyter_text>- _We use Scatter plot for visualization and represent the dataset._<jupyter_code>import matplotlib.pyplot as plt
import numpy as np
# Set the size of the plot
plt.figure(figsize=(14,7))
 
# Create a colormap
colormap = np.array(['red', 'lime', 'black'])
 
# Plot CDR
plt.subplot(1, 2, 1)
plt.scatter(pddf_pred.Calling_NumberIndex, pddf_pred.Called_NumberIndex, c=pddf_pred.prediction)
plt.title('CallDetailRecord')
plt.show()<jupyter_output><empty_output><jupyter_text>Evaluation<jupyter_code>from pyspark.ml.evaluation import MulticlassClassificationEvaluator

evaluator = MulticlassClassificationEvaluator(labelCol="label", predictionCol="prediction",
                                              metricName="accuracy")
accuracy = evaluator.evaluate(predictiondf)
print(accuracy)<jupyter_output>1.0<jupyter_text>Confusion Matrix<jupyter_code>from sklearn.metrics import confusion_matrix
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sn

outdataframe = predictiondf.select("prediction", "label")
pandadf = outdataframe.toPandas()
npmat = pandadf.values
labels = npmat[:,0]
predicted_label = npmat[:,1]

cnf_matrix = confusion_matrix(labels, predicted_label)

import numpy as np

def plot_confusion_matrix(cm,
                          target_names,
                          title='Confusion matrix',
                          cmap=None,
                          normalize=True):

    import matplotlib.pyplot as plt
    import numpy as np
    import itertools

    accuracy = np.trace(cm) / float(np.sum(cm))
    misclass = 1 - accuracy

    if cmap is None:
        cmap = plt.get_cmap('Blues')

    plt.figure(figsize=(8, 6))
    plt.imshow(cm, interpolation='nearest', cmap=cmap)
    plt.title(title)
    plt.colorbar()

    if target_names is not None:
        tick_marks = np.arange(len(target_names))
        plt.xticks(tick_marks, target_names, rotation=45)
        plt.yticks(tick_marks, target_names)

    if normalize:
        cm = cm.astype('float') / cm.sum(axis=1)[:, np.newaxis]

    thresh = cm.max() / 1.5 if normalize else cm.max() / 2
    for i, j in itertools.product(range(cm.shape[0]), range(cm.shape[1])):
        if normalize:
            plt.text(j, i, "{:0.4f}".format(cm[i, j]),
                     horizontalalignment="center",
                     color="white" if cm[i, j] > thresh else "black")
        else:
            plt.text(j, i, "{:,}".format(cm[i, j]),
                     horizontalalignment="center",
                     color="white" if cm[i, j] > thresh else "black")


    plt.tight_layout()
    plt.ylabel('label')
    plt.xlabel('Predicted \naccuracy={:0.4f}; misclass={:0.4f}'.format(accuracy, misclass))
    plt.show()
plot_confusion_matrix(cnf_matrix,
                      normalize    = False,
                      target_names = ['Positive', 'Negative'],
                      title        = "Confusion Matrix")
from pyspark.mllib.evaluation import MulticlassMetrics
# Create (prediction, label) pairs
predictionAndLabel = predictiondf.select("prediction", "label").rdd

# Generate confusion matrix
metrics = MulticlassMetrics(predictionAndLabel)
print(metrics.confusionMatrix())<jupyter_output>DenseMatrix([[5469.]])<jupyter_text>Cross Validation<jupyter_code>from pyspark.ml.tuning import ParamGridBuilder, CrossValidator

# Create ParamGrid and Evaluator for Cross Validation
paramGrid = ParamGridBuilder().addGrid(nb.smoothing, [0.0, 0.2, 0.4, 0.6, 0.8, 1.0]).build()
cvEvaluator = MulticlassClassificationEvaluator(metricName="accuracy")
# Run Cross-validation
cv = CrossValidator(estimator=pipeline, estimatorParamMaps=paramGrid, evaluator=cvEvaluator)
cvModel = cv.fit(trainData)
# Make predictions on testData. cvModel uses the bestModel.
cvPredictions = cvModel.transform(testData)
cvPredictions.select("label", "prediction", "probability").show()
# Evaluate bestModel found from Cross Validation
evaluator.evaluate(cvPredictions)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11486486486486487, 'fraction_numerical': 0.04298986486486486, 'mean_word_length': 4.630527817403709, 'pattern_counts': {'":': 1, '<': 50, '<?xml version=': 0, '>': 52, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '12', 'dir': 'jupyter-structured-clean-dedup', 'id': '9141', 'license': 'Apache-2.0', 'n_tokens_mistral': 4077, 'n_tokens_neox': 3719, 'n_words': 867, 'path': 'source/industry/telecom/notebooks/Ml-Telecom-NaiveBayes.ipynb', 'repo_name': 'Devesh929/Telecom-Data-insight-project', 'size': '11840'}