Notes

%spark.pyspark
import matplotlib.pyplot as plt
import pandas as pd
import os
import numpy as np
import scipy
import seaborn as sb
import csv
import scipy.stats as scs
import traceback
import math
from scipy.stats.stats import pearsonr, spearmanr, kendalltau
from pyspark.mllib.stat import Statistics
from pyspark.sql.functions import udf, when
from pyspark.sql.types import FloatType, TimestampType
from pyspark.sql import functions as F


wng_path = os.getcwd()+"/Sample Data-UC1/WNG KPI/"
oss_path = os.getcwd()+"/Sample Data-UC1/OSS KPI/csv/"


def null_replacer(x):
try:
return float(x)
except:
return -1


replace_null_values = udf(lambda x: null_replacer(x)) #Define UDF function to call

exceptions = ["Period_Start_Time", "LNCEL_Name"]


def process_null_values(dataframe):
for column in dataframe.columns:
if column not in exceptions:
dataframe = dataframe.withColumn(column, replace_null_values(column).cast(FloatType()))
else:
pass
return dataframe


def remove_empty_columns(dataframe):
empty_columns = []
dataset_length = dataframe.count()
for column in dataframe.columns:
if column not in exceptions:
column_sum = dataframe.select(F.sum(dataframe[column])).rdd.map(lambda x: x[0]).collect()[0]
if abs(column_sum) == dataset_length:
empty_columns.append(column)
else:
pass
else:
pass

print empty_columns
dataframe_columns = set(dataframe.columns).difference(set(empty_columns))
print "old:", len(dataframe.columns)
dataframe = dataframe.select(*dataframe_columns)
print "new:", len(dataframe.columns)
return dataframe


def join_dataframes(dataframe1, dataframe2):
dataframe1_columns = set(dataframe1.columns)
dataframe2_columns = set(dataframe2.columns)

common = dataframe1_columns.intersection(dataframe2_columns).difference((set(exceptions)))

select_columns = dataframe2_columns.difference(common)
new_dataframe2 = dataframe2.select(*select_columns)
joined_df = dataframe1.join(new_dataframe2, ["Period_Start_Time", "LNCEL_Name"], 'inner')
print "new:", len(joined_df.columns)
return joined_df


df_wng = spark.read.csv(path = wng_path+"/WNG KPI.csv", header = True, inferSchema = True)
df_wng = df_wng.filter("LNCEL_Name IS NOT NULL")
df_wng = process_null_values(df_wng)
df_wng = remove_empty_columns(df_wng)

tdf1 = spark.read.csv(path = oss_path+ "RSLTE001_System_Program1.csv", header = True, inferSchema = True)
tdf2 = spark.read.csv(path = oss_path+ "RSLTE001_System_Program2.csv", header = True, inferSchema = True)
tdf3 = spark.read.csv(oss_path+ "RSLTE001_System_Program3.csv", header = True, inferSchema = True)
df1 = tdf1.union(tdf2).union(tdf3)
df1 = process_null_values(df1)
df1 = remove_empty_columns(df1)

tdf4 = spark.read.csv(path = oss_path + "RSLTE017_PRB1.csv", header = True, inferSchema = True)
tdf5 = spark.read.csv(path = oss_path + "RSLTE017_PRB2.csv", header = True, inferSchema = True)
tdf6 = spark.read.csv(path = oss_path + "RSLTE017_PRB3.csv", header = True, inferSchema = True)
df2 = tdf4.union(tdf5).union(tdf6)
df2 = process_null_values(df2)
df2 = remove_empty_columns(df2)

tdf7 = spark.read.csv(path = oss_path + "RSLTE054_Capacity1.csv", header = True, inferSchema = True)
tdf8 = spark.read.csv(path = oss_path + "RSLTE054_Capacity2.csv", header = True, inferSchema = True)
tdf9 = spark.read.csv(path = oss_path + "RSLTE054_Capacity3.csv", header = True, inferSchema = True)
df3 = tdf7.union(tdf8).union(tdf9)
df3 = process_null_values(df3)
df3 = remove_empty_columns(df3)

tdf10= spark.read.csv(path = oss_path + "RSLTE052_VoLTE1.csv", header = True, inferSchema = True)
tdf11= spark.read.csv(path = oss_path + "RSLTE052_VoLTE2.csv", header = True, inferSchema = True)
tdf12= spark.read.csv(path = oss_path + "RSLTE052_VoLTE3.csv", header = True, inferSchema = True)
df4 = tdf10.union(tdf11).union(tdf12)
df4 = process_null_values(df4)
df4 = remove_empty_columns(df4)


df5 = spark.read.csv(path = oss_path + "RSLTE024_RSSI.csv", header = True, inferSchema = True)
df6 = spark.read.csv(path = oss_path + "RSLTE025_SINR.csv", header = True, inferSchema = True)
df7 = spark.read.csv(path = oss_path + "RSLTE_RSRQ_RSRP_Histogram.csv", header = True, inferSchema = True)

def to_time(x):
arr = x.split(" ")[0].split(".")
dt_str = arr[2]+ "-" + arr[0] + "-" + arr[1]
dt_str = dt_str + " " + x.split(" ")[1]
return dt_str

convert_period_time_format = udf(lambda x: to_time(x)) #Define UDF function to call

df5 = df5.withColumn("Period_Start_Time",convert_period_time_format("Period_Start_Time").cast(TimestampType()))
df6 = df6.withColumn("Period_Start_Time",convert_period_time_format("Period_Start_Time").cast(TimestampType()))
df7 = df7.withColumn("Period_Start_Time",convert_period_time_format("Period_Start_Time").cast(TimestampType()))

df5 = process_null_values(df5)
df5 = remove_empty_columns(df5)

df6 = process_null_values(df6)
df6 = remove_empty_columns(df6)

df7 = process_null_values(df7)
df7 = remove_empty_columns(df7)

joined_dataframe = join_dataframes(df_wng, df1)
joined_dataframe = join_dataframes(joined_dataframe, df2)
joined_dataframe = join_dataframes(joined_dataframe, df3)
joined_dataframe = join_dataframes(joined_dataframe, df4)
joined_dataframe = join_dataframes(joined_dataframe, df5)
joined_dataframe = join_dataframes(joined_dataframe, df6)
joined_dataframe = join_dataframes(joined_dataframe, df7)

joined_dataframe.select("Audio Pkt Loss Rate_WNG Up").show(100)

joined_dataframe.write.option("header", "true").csv("processed_data")