<jupyter_start><jupyter_text>Changing data type<jupyter_code>df = df.withColumn('Quantity', F.col('Quantity').cast('integer'))
df = df.withColumn('UnitPrice', F.col('UnitPrice').cast('double'))

func = udf(lambda x: datetime.strptime(x, '%Y-%m-%d %H:%M:%S'), DateType())
df = df.withColumn('InvoiceDate', func(col('InvoiceDate')))
df.printSchema()
df.describe().show()
df.toPandas().isnull().sum()
df.agg(F.countDistinct(df.CustomerID)).show()
print('There"re', len(df.select('CustomerID').filter( df.CustomerID.isNotNull()).distinct().collect()), ' distinct customers.')
df.write.mode("overwrite").saveAsTable("sql_table1")
min_max_invoicedate = ss.sql('''
select min(InvoiceDate), max(invoiceDate) from sql_table1
''')
min_max_invoicedate.show()
# Calculating total amount by customer & invoice level
invoice_totalamount = ss.sql('''
select Customerid, InvoiceNo, sum(quantity * unitprice) as TransTotalAmount from sql_table1 where customerid is not null group by customerid, invoiceno
''')
# Calculating average transaction spending for each customer
temp2 = invoice_totalamount.groupBy('CustomerID')\
.agg(
    F.countDistinct('InvoiceNo').alias('NumOfTrans2')
    ,F.sum('TransTotalAmount').alias('net_sales')
    )\
.withColumn('MeanTransSpend', F.col('net_sales')/F.col('NumOfTrans2')) 
print((temp2.count(), len(temp2.columns)))
temp2.limit(5).show()<jupyter_output>(4372, 4)
+----------+-----------+------------------+-----------------+
|CustomerID|NumOfTrans2|         net_sales|   MeanTransSpend|
+----------+-----------+------------------+-----------------+
|     15555|         20|            4758.2|           237.91|
|     15574|          4|            702.25|         175.5625|
|     15634|          1|            243.55|           243.55|
|     13610|          9|1115.4300000000003|123.9366666666667|
|     13192|          2|            911.94|           455.97|
+----------+-----------+------------------+-----------------+<jupyter_text>Aggregating the Main Table<jupyter_code>result = df.withColumn('TotalPrice', F.col('Quantity') * F.col('UnitPrice')) \
.filter(   (df.CustomerID.isNotNull()) &  (df.Description.isNotNull()) )\
.groupBy('CustomerID')\
.agg(
    F.max('InvoiceDate').alias('Last_Transaction_Date'),        
     F.max('UnitPrice').alias('MaxProductSpend'),                                                                                   
     F.sum('Quantity').alias('NumofProds'),                                        
     F.countDistinct('InvoiceNo').alias('NumofTrans'), 
     F.countDistinct('InvoiceDate').alias('Frequency'),                                            
     F.countDistinct('StockCode').alias('DistinctItems'),
     F.sum('TotalPrice').alias('TotalSpend')
     )\
.withColumn('DaysSinceTrans', (-1*F.datediff(F.col('Last_Transaction_Date'), \
F.to_date(F.lit(str('2011-12-31'))))).alias('diff'))\
.withColumn('MeanProductSpend', F.col('TotalSpend') / F.col('NumofProds'))
result.limit(5).show()
result.describe().show()
# eliminating those customer with 2 countries
customer_country = ss.sql('''
select distinct Customerid, Country from sql_table1 
where customerid is not null and customerid in 
(select customerid from sql_table1 
group by customerid 
having count(distinct country) = 1)
''')
print((customer_country.count(), len(customer_country.columns)))
output = result.join(temp2, ['CustomerId'], 'left').join(customer_country, ['CustomerId'] ,'left')
print((output.count(), len(output.columns)))
output.limit(3).show()
final = output.select(['CustomerID', 
 'Country',
 'Last_Transaction_Date',
 'MaxProductSpend',
 'NumofProds',
 'NumofTrans',
 'Frequency',
 'DistinctItems',
 'TotalSpend',
 'DaysSinceTrans',
 'MeanProductSpend',
 'MeanTransSpend'])<jupyter_output><empty_output><jupyter_text>Handling Outliers<jupyter_code>col_outlier = ['MaxProductSpend','NumofProds','NumofTrans', 'Frequency','DistinctItems', 'TotalSpend', 'MeanProductSpend','MeanTransSpend'] 
col_outlier2 = []
for col in col_outlier:
    col_outlier2.append(col+'_RemovedOutliers')
outlier_stats2 = output.select(col_outlier).describe().collect()
# outlier_stats2
pd.DataFrame(outlier_stats2, columns = ['Calculation'] + col_outlier)
print(' mean: ',outlier_stats2[1]['MaxProductSpend']
      , '\n stddev: ',outlier_stats2[2]['MaxProductSpend']
      , '\n max: ',outlier_stats2[4]['MaxProductSpend'])
n_stddev = 4
for col in col_outlier:
    min_ = float(outlier_stats2[3][col])
    max_ = float(outlier_stats2[4][col])
    mean = float(outlier_stats2[1][col])
    stddev = float(outlier_stats2[2][col])
    max_limit = int(min(max_,mean + n_stddev*stddev))
    min_limit = int(max(min_,mean - n_stddev*stddev))
    print(col + '\n' +  'min: ' + str(min_) + ' , max: ' + str(max_))
    print('min_limit: ' + str(min_limit) + ' , max_limit: ' + str(max_limit) + '\n')
n_stddev = 4
for col in col_outlier:
    min_ = float(outlier_stats2[3][col])
    max_ = float(outlier_stats2[4][col])
    mean = float(outlier_stats2[1][col])
    stddev = float(outlier_stats2[2][col])
    max_limit = min(max_,mean + n_stddev*stddev)
    min_limit = max(min_,mean - n_stddev*stddev)
    final = final.withColumn(col+ '_RemovedOutliers', F.when(F.col(col)>max_limit, max_limit)\
                             .when(F.col(col)<min_limit, min_limit)\
                             .otherwise(F.col(col))).drop(col)
final.limit(5).show()
ID_cols = ['CustomerID']
cat_cols = ['Country']
non_cat_cols = ['MaxProductSpend_RemovedOutliers'
, 'NumofProds_RemovedOutliers'
, 'NumofTrans_RemovedOutliers'
, 'Frequency_RemovedOutliers'
, 'DistinctItems_RemovedOutliers'
, 'TotalSpend_RemovedOutliers'
, 'MeanProductSpend_RemovedOutliers'
, 'MeanTransSpend_RemovedOutliers']
column_string = ', '.join(ID_cols + cat_cols + non_cat_cols)
column_string
final.write.mode("overwrite").saveAsTable('member_add_features')
ss.sql('''
select * from member_add_features
''').limit(3).show()
df_features_query = ('''
    SELECT {0}
    FROM member_add_features 
    where NumofProds_RemovedOutliers > 0 
    and TotalSpend_RemovedOutliers > 0 
    and MeanProductSpend_RemovedOutliers > 0 
    and MeanTransSpend_RemovedOutliers > 0''').format(column_string)

df_features = ss.sql(df_features_query)
df_features.toPandas().isnull().sum()
df_features = df_features.withColumn('Country'                            ,
    F.when(df_features['Country'].isNull(),'Unknown').
    otherwise(df_features['Country']))
df_features.toPandas().isnull().sum()<jupyter_output><empty_output><jupyter_text>After Data preprocessing (step 35 - 51), we apply Elbow Method to select the k value<jupyter_code>from pylab import rcParams
rcParams['figure.figsize'] = 14, 7
plt.plot(k, wcss)
plt.title('Elbow Method')
plt.xlabel('Number of clusters')
plt.ylabel('WCSS')
xi = list(k)
plt.xticks(xi)
plt.show()
## We select the number of clusters where the change in WCSS begins to level off, let's pick 8.
k = 8
for i in range(1,6):
    kmeanstest = KMeans(featuresCol='features', k=k, maxIter=1000, tol=1e-4).setSeed(i)
    model_test = kmeanstest.fit(data)
    best_results = model_test.transform(data)
    best_results.groupBy('prediction').count().orderBy('prediction').show()
seed = 4

kmeans2 = KMeans(featuresCol='features', k=k, maxIter=1000, tol=1e-4).setSeed(seed)
model2 = kmeans2.fit(data)
results = model2.transform(data)
results.groupBy('prediction').count().orderBy('prediction').show()
pca = PCA(k=3, inputCol="features", outputCol="pcaFeatures")
model_pca = pca.fit(results)
print(model_pca.explainedVariance)

results_pca = model_pca.transform(results)

vector_udf = F.udf(lambda vector: vector.toArray().tolist(),ArrayType(DoubleType()))

results_df = results_pca.withColumn('PCA', vector_udf('pcaFeatures'))\
                         .withColumn('PCA_1', F.col('PCA')[0])\
                         .withColumn('PCA_2', F.col('PCA')[1])\
                         .withColumn('PCA_3', F.col('PCA')[2])
sum(model_pca.explainedVariance)
results_df.limit(3).toPandas()
cust_cluster_df = results_df.withColumn('ClusterID', F.col('prediction')+1)\
                               .drop('prediction')
df_outcome = cust_cluster_df.select('CustomerID','ClusterID','PCA_1','PCA_2','PCA_3').toPandas()
df_outcome = df_outcome.sort_values(by = 'ClusterID')
df_outcome.head()
df_outcome.CustomerID.nunique()
df_outcome.ClusterID.value_counts()#.to_csv('Cluster_size.csv')<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1783712888994873, 'fraction_numerical': 0.025515679027065696, 'mean_word_length': 4.004773269689737, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 5, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '4', 'dir': 'jupyter-structured-clean-dedup', 'id': '309646', 'license': 'MIT', 'n_tokens_mistral': 3023, 'n_tokens_neox': 2817, 'n_words': 436, 'path': 'PySpark/Pyspark_KMeans.ipynb', 'repo_name': 'rscriptshare/Scripts', 'size': '8387'}