<jupyter_start><jupyter_text>Imports<jupyter_code>from pyspark.sql import SparkSession
import pyspark.sql.functions as F
from pathlib import Path

from pyspark.ml.feature import OneHotEncoder, StringIndexer, VectorAssembler
from pyspark.ml import Pipeline
from pyspark.ml.clustering import KMeans
from pyspark.ml.evaluation import ClusteringEvaluator
DATA_PATH = Path('data/')
!ls {DATA_PATH}
spark = (SparkSession
         .builder
         .appName('Structured Streaming')
         .getOrCreate())
spark<jupyter_output><empty_output><jupyter_text>Structured Streaming<jupyter_code>static_df = (spark
             .read
             .format('csv')
             .option('inferschema', 'true')
             .option('header', 'true')
             .load(str(DATA_PATH / 'retail-data/by-day/*.csv')))
static_schema = static_df.schema
static_schema
static_df.createOrReplaceTempView('retail_data')
static_df.printSchema()
static_df.selectExpr('CustomerID', '(Quantity * UnitPrice) as total_cost', 'InvoiceDate').show()
(static_df
 .selectExpr('CustomerID', '(Quantity * UnitPrice) as total_cost', 'InvoiceDate')
 .groupBy('CustomerID', F.window(F.col('InvoiceDate'), '1 day'))
 .sum('total_cost')
 .show(5))
streaming_df = (spark
                .readStream.format('csv')
                .schema(static_schema)
                .option('maxFilesPerTrigger', 1)
                .option('header', 'true')
                .load(str(DATA_PATH / 'retail-data/by-day/*.csv')))
streaming_df.isStreaming
purchase_by_customer_by_hour = (streaming_df
                                .selectExpr('CustomerID', '(Quantity * UnitPrice) as total_cost', 'InvoiceDate')
                                .groupBy('CustomerID', F.window(F.col('InvoiceDate'), '1 day'))
                                .sum('total_cost'))
(purchase_by_customer_by_hour
 .writeStream
 .format('memory')
 .queryName('customer_purchases')
 .outputMode('complete')
 .start())
spark.sql("SELECT * FROM customer_purchases ORDER BY 3 DESC").show(5)<jupyter_output>+----------+--------------------+------------------+
|CustomerID|              window|   sum(total_cost)|
+----------+--------------------+------------------+
|   12415.0|[2011-03-02 18:00...|          16558.14|
|   15769.0|[2011-03-16 19:00...|           10065.0|
|      null|[2011-03-16 19:00...| 7876.000000000018|
|   12435.0|[2011-03-16 19:00...|3978.9899999999993|
|      null|[2011-03-02 18:00...| 3538.750000000001|
+----------+--------------------+------------------+
only showing top 5 rows<jupyter_text>Machine Learning<jupyter_code>static_df.printSchema()
prep_df = (static_df
          .na.fill(0)
          .withColumn('day_of_week', F.date_format(F.col('InvoiceDate'), 'EEEE'))
          .coalesce(5))
prep_df.explain()
prep_df.show()
train_df = prep_df.where("InvoiceDate < '2011-07-01'")
test_df = prep_df.where("InvoiceDate >= '2011-07-01'")
train_df.count(), test_df.count()
indexer = (StringIndexer()
           .setInputCol('day_of_week')
           .setOutputCol('day_of_week_index'))

encoder = (OneHotEncoder()
           .setInputCol('day_of_week_index')
           .setOutputCol('day_of_week_encoded'))

vector_assembler = (VectorAssembler()
                    .setInputCols(['UnitPrice', 'Quantity', 'day_of_week_encoded'])
                    .setOutputCol('features'))
           
tfms_pipeline = (Pipeline()
                 .setStages([indexer, encoder, vector_assembler]))
fitted_pipeline = tfms_pipeline.fit(train_df)
transformed_train_df = fitted_pipeline.transform(train_df)
transformed_test_df = fitted_pipeline.transform(test_df)
transformed_train_df.cache()
transformed_train_df.show()
kmeans = KMeans().setK(20).setSeed(1)
kmeans_model = kmeans.fit(transformed_train_df)
transformed_train_df.take(1)
ClusteringEvaluator().evaluate(kmeans_model.transform(transformed_train_df))
ClusteringEvaluator().evaluate(kmeans_model.transform(transformed_test_df))
kmeans_model.clusterCenters()
train_df.groupBy('day_of_week').count().show()
kmeans_model.transform(transformed_train_df).show(3)<jupyter_output>+---------+---------+--------------------+--------+-------------------+---------+----------+--------------+-----------+-----------------+-------------------+--------------------+----------+
|InvoiceNo|StockCode|         Description|Quantity|        InvoiceDate|UnitPrice|CustomerID|       Country|day_of_week|day_of_week_index|day_of_week_encoded|            features|prediction|
+---------+---------+--------------------+--------+-------------------+---------+----------+--------------+-----------+-----------------+-------------------+--------------------+----------+
|   537226|    22811|SET OF 6 T-LIGHTS...|       6|2010-12-06 08:34:00|     2.95|   15987.0|United Kingdom|     Monday|              2.0|      (5,[2],[1.0])|(7,[0,1,4],[2.95,...|         0|
|   537226|    21713|CITRONELLA CANDLE...|       8|2010-12-06 08:34:00|      2.1|   15987.0|United Kingdom|     Monday|              2.0|      (5,[2],[1.0])|(7,[0,1,4],[2.1,8...|         0|
|   537226|    22927|GREEN GIANT GARDE...|       2[...]
StarcoderdataJupyterStructuredCleanDedup